第三课：HTTP、REST 与异步网络请求

这节课你会学什么?

Agent 的很多能力最终都要通过网络完成：

  • 调用模型 API
  • 调用搜索接口
  • 访问数据库服务
  • 操作浏览器
  • 和前端通信
  • 调用企业内部系统

所以这一课要学习：

  • client 和 server
  • request 和 response
  • URL、path、query、body、header
  • GET / POST / PUT / DELETE
  • HTTP 状态码
  • JSON API
  • token / API key
  • timeout / retry
  • REST API
  • WebSocket
  • 流式返回
  • 异步 HTTP 请求

最后我们会做一个小项目：

搭建一个简单的 Agent API 服务，并用异步客户端调用它。

1. 什么是网络通信

网络通信可以理解成：

一个程序向另一个程序发送请求，并等待对方返回结果。

例如：

  • 前端向后端请求数据
  • Agent 向搜索服务请求搜索结果
  • Agent 向模型服务发送 prompt
  • 后端向数据库服务查询数据

网络通信通常包含两个角色：

  • Client（客户端）
    发起请求的一方
  • Server（服务器）
    接收请求并返回结果的一方

例如：

浏览器          ->          后端服务器
Agent 程序      ->          模型服务器
Python 客户端   ->          搜索 API

客户端不一定是浏览器，任何能够发起请求的程序都可以是客户端。

2. 什么是 HTTP

HTTP（HyperText Transfer Protocol，超文本传输协议） 是网络通信中最常见的一套规则。

它规定了：

  • 请求应该怎么写
  • 响应应该怎么写
  • 方法如何表示
  • 状态码如何表示
  • 数据如何传输

HTTP 最基本的工作模式是：

Client 发送 Request
       ↓
Server 处理请求
       ↓
Server 返回 Response

例如：

Agent -> 请求模型 API
模型服务 -> 返回模型结果

3. 什么是 request 和 response

3.1 Request（请求）

Request 是客户端发给服务器的数据。

一个 HTTP 请求通常包括：

  • URL
  • Method
  • Header
  • Body

例如：

POST /chat HTTP/1.1
Host: example.com
Authorization: Bearer your-token
Content-Type: application/json

{
  "task": "summarize this paper"
}

3.2 Response（响应）

Response 是服务器返回给客户端的数据。

一个 HTTP 响应通常包括：

  • 状态码
  • Header
  • Body

例如：

HTTP/1.1 200 OK
Content-Type: application/json

{
  "message": "summary result"
}

4. URL 的组成

URL（Uniform Resource Locator，统一资源定位符）就是网络资源的地址。

例如：

https://example.com:443/api/tasks?page=2&limit=10

可以拆成：

https://example.com:443/api/tasks?page=2&limit=10
│      │           │          │
│      │           │          └── query
│      │           └── path
│      └── host 和端口
└── 协议

4.1 Scheme（协议）

https://

表示使用 HTTPS 协议。

  • http://：普通 HTTP
  • https://：加密的 HTTP

4.2 Host（主机）

example.com

表示服务器的域名。

开发时也经常看到：

127.0.0.1
localhost

它们通常表示本机。

4.3 Port（端口）

:8000

端口表示服务器上具体监听哪个网络入口。

4.4 Path（路径）

/api/tasks

路径表示访问哪个资源。

4.5 Query（查询参数）

?page=2&limit=10

查询参数通常用于筛选、分页、排序等。

5. Path、Query、Body、Header

这四个概念非常重要。

5.1 Path 参数

Path 表示 URL 路径的一部分。

GET /tasks/123

这里的 123 就是任务 ID。

可以理解成：

获取 ID 为 123 的任务

5.2 Query 参数

Query 是 URL 后面的查询参数。

GET /tasks?page=2&limit=10

表示：

  • 第 2 页
  • 每页 10 条

5.3 Body 请求体

Body 是请求中携带的主要数据。

{
  "task": "summarize paper",
  "max_steps": 5
}

通常 POST、PUT 请求会携带 body。

5.4 Header 请求头

Header 是请求的附加信息。

常见 Header：

Content-Type: application/json
Authorization: Bearer your-token
User-Agent: my-agent

例如：

  • Content-Type
    表示请求体是什么格式
  • Authorization
    表示身份认证信息
  • User-Agent
    表示客户端身份

6. HTTP 方法

6.1 GET

GET 用来获取数据。

GET /tasks

含义：

获取任务列表

6.2 POST

POST 用来提交数据或创建资源。

POST /tasks

请求体：

{
  "task": "research AI agents"
}

6.3 PUT

PUT 通常用来整体更新资源。

PUT /tasks/123

6.4 DELETE

DELETE 用来删除资源。

DELETE /tasks/123

一个简单记忆方式：

方法 常见用途
GET 获取
POST 创建或提交
PUT 更新
DELETE 删除

7. HTTP 状态码

状态码是服务器告诉客户端“这次请求结果如何”的数字。

7.1 成功类

200 OK

表示请求成功。

201 Created

表示资源创建成功。

204 No Content

表示成功，但没有返回内容。

7.2 客户端错误

400 Bad Request

表示请求格式不正确。

401 Unauthorized

表示没有通过身份认证，通常是 token 缺失或错误。

403 Forbidden

表示身份可能正确，但没有权限。

404 Not Found

表示资源不存在。

7.3 服务端错误

500 Internal Server Error

表示服务器内部发生错误。

503 Service Unavailable

表示服务暂时不可用。

在 Agent 中，你不能只看返回内容，还要检查状态码。

8. 什么是 JSON API

JSON API 是一种使用 JSON 传递请求和响应数据的接口。

例如请求：

{
  "task": "summarize this paper",
  "language": "zh"
}

返回：

{
  "result": "这是一篇关于动态组件系统的论文",
  "status": "done"
}

JSON 很适合 Agent，因为：

  • 结构清晰
  • Python 容易解析
  • 模型容易生成
  • 工具参数可以用 JSON 表示

请求头一般要写：

Content-Type: application/json

它告诉服务器：

请求体里的内容是 JSON。

9. 使用 httpx 发起同步请求

httpx 是 Python 的 HTTP 客户端库。

安装：

pip install httpx

9.1 GET 请求

import httpx

response = httpx.get("https://www.baidu.com/")

print("状态码:", response.status_code)
print("响应文本:", response.text)

解释：

  • httpx.get(...)
    发起 GET 请求
  • response.status_code
    读取状态码
  • response.text
    读取文本内容

9.2 读取 JSON

import httpx

response = httpx.get("https://httpbin.org/get")

if response.status_code == 200:
    data = response.json()
    print(data)
else:
    print("请求失败:", response.status_code)

response.json() 会把 JSON 响应转换成 Python 字典或列表。

10. 使用 httpx 发送 POST 请求

import httpx

payload = {
    "task": "summarize this paper",
    "language": "zh",
}

response = httpx.post(
    "https://httpbin.org/post",
    json=payload,
)

print("状态码:", response.status_code)
print("返回内容:", response.json())

解释：

  • payload
    表示要发送的数据
  • json=payload
    表示让 httpx 自动把字典转换成 JSON
  • httpx.post(...)
    发送 POST 请求

11. Path、Query、Header 的代码写法

import httpx

task_id = 123

params = {
    "page": 2,
    "limit": 10,
}

headers = {
    "X-Client-Name": "my-agent",
}

response = httpx.get(
    f"https://httpbin.org/anything/tasks/{task_id}",
    params=params,
    headers=headers,
)

print(response.url)
print(response.json())

这里：

  • f"...{task_id}"：把变量放入 Path
  • params=params：发送 Query 参数
  • headers=headers：发送 Header

最终 URL 类似：

https://httpbin.org/anything/tasks/123?page=2&limit=10

12. 异步 HTTP 请求

Agent 经常需要同时调用多个网络服务，所以异步 HTTP 很重要。

httpx 提供了 AsyncClient。

import asyncio
import httpx


async def fetch_data():
    """
    异步获取一个网页接口的响应。
    """
    async with httpx.AsyncClient() as client:
        response = await client.get("https://httpbin.org/get")
        return response.json()


async def main():
    data = await fetch_data()
    print(data)


if __name__ == "__main__":
    asyncio.run(main())

解释：

  • AsyncClient
    异步 HTTP 客户端
  • async with
    异步上下文管理器，用来正确管理网络连接
  • await client.get(...)
    异步发送请求
  • asyncio.run(main())
    启动事件循环

13. 为什么要复用 HTTP Client

下面这种写法每次都重新创建客户端：

async def fetch_a():
    async with httpx.AsyncClient() as client:
        return await client.get("https://example.com/a")


async def fetch_b():
    async with httpx.AsyncClient() as client:
        return await client.get("https://example.com/b")

如果请求很多，会浪费连接资源。

更好的方式是复用一个客户端：

import httpx


async def fetch_all():
    async with httpx.AsyncClient() as client:
        response_a = await client.get("https://example.com/a")
        response_b = await client.get("https://example.com/b")

        return response_a, response_b

这样可以：

  • 复用连接
  • 减少资源开销
  • 更容易统一设置超时和 Header

14. 并发发送多个请求

如果多个请求彼此独立，可以并发发送。

import asyncio
import httpx


async def fetch(client: httpx.AsyncClient, url: str) -> dict:
    """
    使用传入的 HTTP 客户端请求指定 URL。
    """
    response = await client.get(url)
    response.raise_for_status()
    # 如果状态码是 4xx 或 5xx，这里会抛出异常

    return response.json()


async def main():
    urls = [
        "https://httpbin.org/anything/1",
        "https://httpbin.org/anything/2",
        "https://httpbin.org/anything/3",
    ]

    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(
            *(fetch(client, url) for url in urls)
        )

    for result in results:
        print(result)


if __name__ == "__main__":
    asyncio.run(main())

这里的重点：

  • asyncio.gather(...)
    并发执行多个请求
  • *
    把列表中的多个协程展开成多个参数
  • raise_for_status()
    如果请求失败，就抛出异常

15. timeout（超时）

timeout（超时） 表示：

请求最多等待多长时间。

网络请求不能无限等待，否则一个卡住的工具可能让整个 Agent 永远停住。

import asyncio
import httpx


async def fetch_with_timeout():
    timeout = httpx.Timeout(5.0)

    async with httpx.AsyncClient(timeout=timeout) as client:
        response = await client.get("https://httpbin.org/delay/2")
        return response.json()


async def main():
    try:
        result = await fetch_with_timeout()
        print(result)
    except httpx.TimeoutException:
        print("请求超时")


if __name__ == "__main__":
    asyncio.run(main())

这里：

  • httpx.Timeout(5.0)
    表示请求最多等待 5 秒
  • TimeoutException
    表示请求超时

Agent 里建议给不同工具设置不同超时：

  • 普通 API：几秒
  • 搜索：十几秒
  • 浏览器：几十秒
  • 长任务：更长，但要支持取消

16. retry（重试）

retry（重试） 表示：

请求失败后，再尝试几次。

网络请求偶尔失败并不一定表示服务彻底坏了，可能只是：

  • 网络抖动
  • 临时超时
  • 服务瞬时繁忙

16.1 最简单的重试

import asyncio
import httpx


async def fetch_with_retry(
    client: httpx.AsyncClient,
    url: str,
    retries: int = 3,
) -> dict:
    """
    请求失败后最多重试 retries 次。
    """
    for attempt in range(1, retries + 1):
        try:
            response = await client.get(url)
            response.raise_for_status()
            return response.json()

        except (httpx.TimeoutException, httpx.HTTPError) as e:
            print(f"第 {attempt} 次请求失败: {e}")

            if attempt == retries:
                raise

            # 指数退避（exponential backoff）：
            # 第一次等 1 秒，第二次等 2 秒，第三次等 4 秒
            wait_seconds = 2 ** (attempt - 1)
            await asyncio.sleep(wait_seconds)

    raise RuntimeError("unreachable")

这里的“指数退避”可以减少对故障服务的连续冲击。

16.2 重试要注意什么

不是所有请求都适合重试。

通常：

  • GET 比较适合重试
  • 查询类请求比较适合重试
  • POST 可能会重复创建资源，要谨慎
  • 扣款、发送消息等操作不能随意重试

因为重试可能造成重复操作。

17. token 和 API Key

很多 API 需要身份认证。

API Key 可以理解成访问 API 的密码或钥匙。

错误做法：

api_key = "sk-xxxxxxxx"

不要把真正的密钥直接写进代码，也不要提交到 GitHub。

更好的方式是使用环境变量：

import os

api_key = os.environ.get("MY_API_KEY")

if not api_key:
    raise RuntimeError("MY_API_KEY is not configured")

设置环境变量：

Windows PowerShell:

$env:MY_API_KEY="your-api-key"

Linux / macOS:

export MY_API_KEY="your-api-key"

17.1 Bearer Token

很多 API 使用这种 Header：

Authorization: Bearer your-token

Python 写法：

import os
import httpx

api_key = os.environ["MY_API_KEY"]

headers = {
    "Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json",
}

response = httpx.get(
    "https://example.com/api/data",
    headers=headers,
)

解释：

  • Authorization
    表示身份认证信息
  • Bearer
    表示使用持有者令牌认证
  • os.environ["MY_API_KEY"]
    如果环境变量不存在，会直接抛出异常

18. REST API 是什么

REST（Representational State Transfer） 是一种常见的 Web API 设计风格。

REST 通常遵循这些思想：

  • URL 表示资源
  • HTTP 方法表示动作
  • 使用 JSON 传输数据
  • 每个请求尽量独立
  • 使用状态码表达结果

例如任务系统：

GET    /tasks          获取任务列表
POST   /tasks          创建任务
GET    /tasks/123      获取任务 123
PUT    /tasks/123      更新任务 123
DELETE /tasks/123      删除任务 123

Agent 后端也可以采用类似设计：

POST /chat
POST /tasks
GET  /tasks/{task_id}
POST /tools/{tool_name}
GET  /health

REST 的优点是：

  • 容易理解
  • 容易调试
  • 前后端分离清晰
  • 适合大多数普通请求

19. 用 FastAPI 创建 REST API

安装：

pip install fastapi uvicorn

创建文件 server.py：

from fastapi import FastAPI
from pydantic import BaseModel


app = FastAPI()


class TaskRequest(BaseModel):
    """
    客户端提交任务时需要发送的数据结构。
    """
    task: str


class TaskResponse(BaseModel):
    """
    服务器返回的数据结构。
    """
    task: str
    status: str


@app.get("/health")
async def health():
    """
    健康检查接口。
    """
    return {"status": "ok"}


@app.post("/tasks", response_model=TaskResponse)
async def create_task(req: TaskRequest):
    """
    创建一个新任务。
    """
    return TaskResponse(
        task=req.task,
        status="accepted",
    )

运行：

uvicorn server:app --reload

测试：

http://127.0.0.1:8000/health

FastAPI 还会自动生成接口文档：

http://127.0.0.1:8000/docs

20. 用异步客户端调用 REST API

创建文件 client.py：

import asyncio
import httpx


BASE_URL = "http://127.0.0.1:8000"


async def create_task(task: str) -> dict:
    """
    调用后端的 POST /tasks 接口。
    """
    payload = {
        "task": task,
    }

    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{BASE_URL}/tasks",
            json=payload,
        )

        response.raise_for_status()
        return response.json()


async def main():
    result = await create_task("summarize an AI paper")
    print(result)


if __name__ == "__main__":
    asyncio.run(main())

运行顺序：

  1. 先启动服务端：
uvicorn server:app --reload
  1. 再打开另一个终端，运行客户端：
python client.py

这就是一个完整的：

Python Client -> HTTP Request -> FastAPI Server -> JSON Response

21. WebSocket 是什么

WebSocket 是一种可以长期保持连接的通信协议。

普通 HTTP 通常是：

客户端请求
服务器响应
连接结束

WebSocket 更像是：

客户端和服务器建立连接
双方可以持续发送消息
连接保持一段时间

它适合：

  • 聊天
  • 实时进度
  • Agent 状态更新
  • 日志实时推送
  • 流式输出
  • 多人协作

22. HTTP 和 WebSocket 的区别

特性 HTTP WebSocket
通信方式 请求-响应 双向持续通信
连接 通常一次请求一次响应 长时间保持
适合 普通 API 实时交互
Agent 场景 提交任务、查询结果 实时显示思考/工具进度
复杂度 较低 较高

普通的模型调用可以使用 HTTP。

如果你想让前端实时看到：

正在搜索
正在读取网页
正在调用模型
正在生成报告

就可以考虑 WebSocket 或流式 HTTP。

23. FastAPI WebSocket 服务端

安装：

pip install websockets

创建 websocket_server.py：

import asyncio
from fastapi import FastAPI, WebSocket


app = FastAPI()


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    """
    WebSocket 连接处理函数。

    连接建立后，服务器会持续等待客户端消息。
    """
    await websocket.accept()
    # accept 表示接受客户端的 WebSocket 连接

    try:
        while True:
            message = await websocket.receive_text()
            # receive_text 等待客户端发送文本消息

            await websocket.send_text(
                f"server received: {message}"
            )
            # send_text 把消息发送回客户端

            await asyncio.sleep(0.5)
            await websocket.send_text("agent is working...")

    except Exception:
        # 客户端断开后，退出处理函数
        print("WebSocket connection closed")

运行：

uvicorn websocket_server:app --reload

24. WebSocket 客户端

创建 websocket_client.py：

import asyncio
import websockets


async def main():
    """
    连接 WebSocket 服务，并发送一条消息。
    """
    async with websockets.connect(
        "ws://127.0.0.1:8000/ws"
    ) as websocket:

        await websocket.send("start agent task")
        # send 发送消息给服务器

        for _ in range(2):
            message = await websocket.recv()
            # recv 等待服务器发送消息
            print("received:", message)


if __name__ == "__main__":
    asyncio.run(main())

这里：

  • ws://
    是普通 WebSocket
  • wss://
    是加密 WebSocket
  • send(...)
    发送消息
  • recv()
    接收消息

25. 什么是流式返回

流式返回（streaming response） 是指：

服务器不用等全部结果生成完，而是一小段一小段地返回。

普通返回：

等待 10 秒
一次性返回完整答案

流式返回：

返回第 1 段
返回第 2 段
返回第 3 段
...

模型生成文字时，经常使用流式返回，因为用户可以更早看到结果。

26. FastAPI 流式响应

创建 stream_server.py：

import asyncio
from fastapi import FastAPI
from fastapi.responses import StreamingResponse


app = FastAPI()


async def generate_text():
    """
    一个异步生成器。

    yield 不会一次性返回全部结果，
    而是每次产生一小段内容。
    """
    parts = [
        "Agent ",
        "正在 ",
        "搜索资料。",
        "搜索完成。",
        "正在生成总结。",
    ]

    for part in parts:
        yield part
        await asyncio.sleep(0.5)


@app.get("/stream")
async def stream():
    """
    返回一个流式响应。
    """
    return StreamingResponse(
        generate_text(),
        media_type="text/plain",
    )

解释：

  • yield
    每次产生一小段结果
  • 异步生成器
    指可以使用 async def 和 yield 的函数
  • StreamingResponse
    FastAPI 提供的流式响应对象
  • media_type
    表示返回内容的类型

运行：

uvicorn stream_server:app --reload

27. 异步客户端读取流式结果

import asyncio
import httpx


async def read_stream():
    async with httpx.AsyncClient() as client:
        async with client.stream(
            "GET",
            "http://127.0.0.1:8000/stream",
        ) as response:

            response.raise_for_status()

            async for chunk in response.aiter_text():
                print(chunk, end="", flush=True)


if __name__ == "__main__":
    asyncio.run(read_stream())

解释：

  • client.stream(...)
    表示以流式方式请求
  • async for
    逐段读取异步数据
  • aiter_text()
    按文本块读取响应
  • flush=True
    让内容立刻显示，而不是等缓冲区满了再显示

28. Agent 中的流式返回

Agent 可以把不同阶段实时发送给前端：

开始分析任务
正在调用搜索工具
搜索完成
正在读取论文
正在生成总结
任务完成

这比让用户一直面对空白页面更友好。

常见实现方式：

  • HTTP Streaming
  • Server-Sent Events（SSE）
  • WebSocket

初学阶段可以先掌握：

  • HTTP 普通 JSON 返回
  • HTTP 流式返回
  • WebSocket 基础

29. 小项目：异步 Agent API 服务

现在把这节课的知识串起来。

29.1 服务端：agent_server.py

import asyncio
import logging
from fastapi import FastAPI
from pydantic import BaseModel
from fastapi.responses import StreamingResponse


logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
)

logger = logging.getLogger("agent-server")

app = FastAPI()


class AgentRequest(BaseModel):
    """
    Agent 请求结构。

    客户端必须发送：
    {
        "task": "任务内容"
    }
    """
    task: str


class AgentResponse(BaseModel):
    """
    Agent 普通响应结构。
    """
    task: str
    result: str
    status: str


async def fake_model_call(task: str) -> str:
    """
    模拟一次模型调用。

    真实项目中，这里可以替换为：
    - 模型 API
    - 内部模型网关
    - 本地模型服务
    """
    logger.info("Calling model: %s", task)

    await asyncio.sleep(2)

    return f"模型已经处理任务：{task}"


@app.get("/health")
async def health():
    """
    健康检查接口。
    """
    return {"status": "ok"}


@app.post("/agent", response_model=AgentResponse)
async def run_agent(req: AgentRequest):
    """
    普通 Agent 接口。

    流程：
    1. 接收任务
    2. 调用模型
    3. 返回完整结果
    """
    logger.info("Received task: %s", req.task)

    result = await fake_model_call(req.task)

    return AgentResponse(
        task=req.task,
        result=result,
        status="done",
    )


async def generate_agent_progress(task: str):
    """
    模拟 Agent 的实时进度。
    """
    messages = [
        "任务已接收\n",
        "正在分析任务\n",
        "正在调用搜索工具\n",
        "搜索完成\n",
        "正在生成结果\n",
        f"任务完成：{task}\n",
    ]

    for message in messages:
        yield message
        await asyncio.sleep(0.5)


@app.post("/agent/stream")
async def run_agent_stream(req: AgentRequest):
    """
    流式 Agent 接口。

    它不会一次性返回最终结果，
    而是不断返回当前进度。
    """
    return StreamingResponse(
        generate_agent_progress(req.task),
        media_type="text/plain",
    )

29.2 客户端：agent_client.py

import asyncio
import httpx


BASE_URL = "http://127.0.0.1:8000"


async def run_normal_agent(task: str):
    """
    调用普通 JSON Agent 接口。
    """
    payload = {
        "task": task,
    }

    timeout = httpx.Timeout(10.0)

    async with httpx.AsyncClient(timeout=timeout) as client:
        response = await client.post(
            f"{BASE_URL}/agent",
            json=payload,
        )

        response.raise_for_status()
        return response.json()


async def run_stream_agent(task: str):
    """
    调用流式 Agent 接口，并实时打印进度。
    """
    payload = {
        "task": task,
    }

    async with httpx.AsyncClient(timeout=None) as client:
        async with client.stream(
            "POST",
            f"{BASE_URL}/agent/stream",
            json=payload,
        ) as response:

            response.raise_for_status()

            async for chunk in response.aiter_text():
                print(chunk, end="", flush=True)


async def main():
    """
    主程序同时演示普通调用和流式调用。
    """
    normal_result = await run_normal_agent(
        "总结这篇论文的主要思想"
    )
    print("普通结果:")
    print(normal_result)

    print("\n流式结果:")
    await run_stream_agent(
        "分析动态 Agent 的技术路线"
    )


if __name__ == "__main__":
    asyncio.run(main())

29.3 运行项目

先启动服务端：

uvicorn agent_server:app --reload

再打开另一个终端运行客户端：

python agent_client.py

整体流程：

Python Client
    ↓
HTTP POST /agent
    ↓
FastAPI Server
    ↓
模拟模型调用
    ↓
JSON Response

流式接口流程：

Python Client
    ↓
HTTP POST /agent/stream
    ↓
FastAPI Server
    ↓
持续返回进度
    ↓
客户端实时显示

30. 这节课你应该怎么练

第一次练习

先把服务端和客户端原样跑通。

第二次练习

给请求增加语言参数：

class AgentRequest(BaseModel):
    task: str
    language: str = "zh"

然后让服务端返回：

{
  "task": "summarize paper",
  "language": "zh",
  "result": "..."
}

第三次练习

给普通请求增加重试功能：

  • 第一次请求失败后等待 1 秒
  • 第二次请求失败后等待 2 秒
  • 第三次仍失败就抛出异常

第四次练习

增加工具进度：

正在调用 search
search 调用完成
正在调用 read_file
read_file 调用完成

第五次练习

增加 WebSocket 接口，让客户端可以：

  • 发送任务
  • 接收实时进度
  • 发送取消指令

31. 这节课你真正应该记住的东西

  • Client 是发起请求的一方，Server 是处理请求的一方。
  • Request 是请求，Response 是响应。
  • URL 用来定位资源。
  • Path 表示资源路径。
  • Query 用来传递筛选、分页等参数。
  • Body 用来传递主要数据。
  • Header 用来传递元信息和认证信息。
  • GET 通常用于获取，POST 通常用于提交或创建。
  • 状态码可以快速判断请求是否成功。
  • JSON 是 Agent 和 API 之间常见的数据格式。
  • API Key 是访问服务的凭证，不能直接写进代码。
  • timeout 防止请求无限等待。
  • retry 可以处理临时性失败，但不能无脑重试所有请求。
  • REST 适合普通请求-响应接口。
  • WebSocket 适合长期保持连接和实时双向通信。
  • 流式返回适合模型输出和 Agent 进度展示。
  • 异步 HTTP 适合同时调用多个网络工具。

32. 下一步该学什么？

学完这一课，建议进入下一课：

JSON Schema、Pydantic 与 Tool Router

下一课你会学习：

  • 如何定义工具参数
  • 如何校验工具输入
  • 如何统一工具输出
  • 如何注册多个工具
  • 如何根据工具名分发调用
  • 如何让模型调用工具

到那一步，你的 Agent 就会从：

输入 -> 模型 -> 输出

升级成：

输入 -> 模型决定工具 -> 调用工具 -> 获取结果 -> 继续执行

:::

# 网络 # python # http
Logo
AI硬件创业社区

智能硬件社区聚焦AI智能硬件技术生态，汇聚嵌入式AI、物联网硬件开发者，打造交流分享平台，同步全国赛事资讯、开展 OPC 核心人才招募，助力技术落地与开发者成长。

更多推荐

ESP32-S3 + I2S 麦克风实时音频流静音问题排查：从网络到AGC的谬误溯源

本文分析了ESP32数字麦克风音频采集系统中出现的周期性静音问题。系统通过WiFi UDP传输16kHz/16bit音频数据，经服务器AGC处理后网页播放。关键现象是每200-300ms清晰音频后伴随10-30ms完全静音。经过排除浏览器缓冲不足、JavaScript性能、UDP丢包等错误假设后，发现根源在于Rust服务端的AGC处理函数使用了chunks_exact_mut方法，导致每100ms

avatar AI硬件创业社区
cover

【proteus仿真】基于 STM32 的智能垃圾桶设计（仿真图+程序）

avatar AI硬件创业社区
cover

从一道GESP真题出发：聊聊贪心排序与前缀和优化

avatar AI硬件创业社区