Agent 开发之路

主题

项目 4 · 全栈 Agent 应用 ​

🎯 难度：中级⏱ 预计：2–3 天# FastAPI# 流式# 前端# 取消

终于进入你的主场：浏览器输入任务 → Python Agent 决定是否调用换算工具 → 页面显示工具状态 → 流式显示最终回答。

本版每次提交是独立任务，不包含跨轮记忆。前端用原生 HTML/JS，方便看清协议；以后迁移到 React/Vue 只是组件与状态管理的变化。

架构与准备 ​

text
浏览器 POST /api/agent
  → 后端校验输入
  → 非流式工具决策循环（最多 4 轮）
  → 流式生成最终答案
  → NDJSON 事件：status / tool / token / done / error

NDJSON 就是一行一个 JSON 对象。HTTP 分块边界不等于行边界，前端必须缓冲并按换行切割。我们不用 SSE，避免初学时同时学习 EventSource 和 POST 流。

bash
uv init fullstack-agent --python 3.12
cd fullstack-agent
uv add "openai>=1.68,<2" "pydantic>=2,<3" "python-dotenv>=1,<2" "fastapi>=0.115,<1" "uvicorn>=0.30,<1"

复制 config.py、runner.py、demo_agent.py，创建 .env。本项目只使用 demo_agent 的换算工具，天气是模拟数据，所以这里不引入天气工具。

Python 后端 ​

python
import asyncio
import json
from pathlib import Path
from fastapi import FastAPI, Request
from fastapi.responses import FileResponse, StreamingResponse
from openai import OpenAIError
from pydantic import BaseModel, Field
from config import get_async_client, model_name
from demo_agent import ConvertArgs, convert
from runner import dispatch, tool_spec

app = FastAPI()
SLOTS = asyncio.Semaphore(2)
TOOLS = [tool_spec("convert", "将摄氏度换算为华氏度", ConvertArgs)]
HANDLERS = {"convert": (ConvertArgs, convert)}

class Task(BaseModel):
    message: str = Field(min_length=1, max_length=1_500)


def event(kind: str, **data) -> str:
    return json.dumps({"type": kind, **data}, ensure_ascii=False) + "\n"


@app.get("/")
def index():
    return FileResponse(Path(__file__).resolve().parent / "index.html")


@app.post("/api/agent")
async def agent(body: Task, request: Request):
    async def generate():
        async with SLOTS:
            try:
                async with get_async_client() as client:
                    messages = [
                        {"role": "system", "content": "你是中文助手，温度换算必须使用工具。不得编造工具结果。"},
                        {"role": "user", "content": body.message},
                    ]
                    calls = 0
                    for step in range(4):
                        if await request.is_disconnected():
                            return
                        yield event("status", text=f"正在处理第 {step + 1} 轮")
                        reply = await client.chat.completions.create(
                            model=model_name(), messages=messages,
                            tools=TOOLS, tool_choice="auto", max_tokens=500,
                        )
                        choice = reply.choices[0]
                        msg = choice.message
                        if choice.finish_reason not in ("stop", "tool_calls"):
                            raise RuntimeError("工具决策未正常完成")
                        if not msg.tool_calls:
                            # 决策结束；丢弃这次非流式草稿，接下来重新流式生成最终回答
                            break
                        messages.append(msg.model_dump(exclude_none=True))
                        for call in msg.tool_calls:
                            calls += 1
                            if calls > 4:
                                raise RuntimeError("工具预算已用完")
                            yield event("tool", name=call.function.name)
                            result = dispatch(call.function.name, call.function.arguments, HANDLERS)
                            messages.append({"role": "tool", "tool_call_id": call.id, "content": result})
                    else:
                        raise RuntimeError("模型循环预算已用完")

                    yield event("status", text="正在生成回答")
                    stream = await client.chat.completions.create(
                        model=model_name(), messages=messages, tools=TOOLS,
                        tool_choice="none", stream=True, max_tokens=800,
                    )
                    finish = None
                    try:
                        async for chunk in stream:
                            if await request.is_disconnected():
                                return
                            if not chunk.choices:
                                continue
                            choice = chunk.choices[0]
                            if choice.finish_reason:
                                finish = choice.finish_reason
                            if choice.delta.content:
                                yield event("token", text=choice.delta.content)
                    finally:
                        await stream.close()
                    if finish != "stop":
                        raise RuntimeError("回答中断或被截断")
                    yield event("done")
            except (OpenAIError, ValueError, RuntimeError):
                # 细节应脱敏记录在服务端，不发送 API 请求头或密钥到页面
                yield event("error", text="任务失败，请检查配置、网络或任务预算")
    return StreamingResponse(
        generate(), media_type="application/x-ndjson",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

静态前端 ​

html
<!doctype html>
<html lang="zh-CN">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width,initial-scale=1">
  <title>我的第一个 Agent</title>
  <style>
    * { box-sizing: border-box; }
    body { margin: 0; background: #0b0b14; color: #eceafa; font-family: system-ui; }
    main { max-width: 760px; margin: 60px auto; padding: 24px; }
    h1 { color: #b9a0ff; }
    textarea, pre { width: 100%; background: #17172b; color: inherit; border: 1px solid #35334f; border-radius: 12px; padding: 16px; }
    textarea { min-height: 100px; resize: vertical; font: inherit; }
    pre { white-space: pre-wrap; overflow-wrap: anywhere; min-height: 150px; font: inherit; line-height: 1.8; }
    button { padding: 10px 18px; border: 0; border-radius: 9px; margin: 12px 8px 0 0; cursor: pointer; background: #8756f6; color: white; }
    button:disabled { opacity: .4; cursor: default; }
    #status { color: #72d8ee; }
  </style>
</head>
<body>
<main>
  <h1>Agent 工作台</h1>
  <p>每次提交是独立任务。试试把 22°C 换算成华氏度。</p>
  <form id="form">
    <label for="question">你的任务</label>
    <textarea id="question" maxlength="1500" required>把 22 摄氏度换算成华氏度，并解释公式。</textarea>
    <button id="send">开始</button>
    <button id="cancel" type="button" disabled>停止</button>
  </form>
  <p id="status" role="status"></p>
  <pre id="answer" aria-label="Agent 回答"></pre>
</main>
<script type="module">
const form = document.querySelector('#form')
const question = document.querySelector('#question')
const status = document.querySelector('#status')
const answer = document.querySelector('#answer')
const send = document.querySelector('#send')
const cancel = document.querySelector('#cancel')
let controller
cancel.onclick = () => controller?.abort()

form.onsubmit = async (e) => {
  e.preventDefault()
  if (!question.value.trim()) return
  controller = new AbortController()
  send.disabled = true
  cancel.disabled = false
  answer.textContent = ''
  status.textContent = '连接中…'
  let completed = false
  let failed = false
  let reader
  const consume = (line) => {
    if (!line.trim()) return
    const item = JSON.parse(line)
    if (item.type === 'token') answer.textContent += item.text
    if (item.type === 'status') status.textContent = item.text
    if (item.type === 'tool') status.textContent = `正在执行工具：${item.name}`
    if (item.type === 'done') { completed = true; status.textContent = '完成' }
    if (item.type === 'error') { failed = true; status.textContent = item.text }
  }
  try {
    const res = await fetch('/api/agent', {
      method: 'POST', headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ message: question.value }), signal: controller.signal,
    })
    if (!res.ok || !res.body) throw new Error(`请求失败（HTTP ${res.status}）`)
    reader = res.body.getReader()
    const decoder = new TextDecoder()
    let buffer = ''
    while (true) {
      const { value, done } = await reader.read()
      buffer += done ? decoder.decode() : decoder.decode(value, { stream: true })
      const lines = buffer.split('\n')
      buffer = lines.pop()
      lines.forEach(consume)
      if (done) break
    }
    if (buffer.trim()) consume(buffer)
    if (!completed && !failed) throw new Error('连接结束但任务未完成')
  } catch (error) {
    status.textContent = error.name === 'AbortError' ? '已停止，当前回答可能不完整' : error.message
  } finally {
    if (reader) {
      await reader.cancel().catch(() => {})
      reader.releaseLock()
    }
    send.disabled = false
    cancel.disabled = true
  }
}
</script>
</body>
</html>

运行 ​

bash
uv run uvicorn app:app --host 127.0.0.1 --port 8000 --reload

打开 http://127.0.0.1:8000。前端由同一个 Python 服务提供，所以不需要额外配置 CORS。这里启动的是练习应用，与文档站不同。

关键设计选择 ​

  • 工具决策先非流式：不需要拼接半个工具参数 JSON。
  • 最终回答另开一次流式调用：清晰易懂，但多一次调用和费用；后续可以优化。
  • 浏览器只请求后端：API Key 不进入网页。
  • 用 textContent 展示内容：模型输出不能作为 HTML 执行；渲染 Markdown 时另做安全处理。
  • AbortController 中断浏览器请求；后端生成器取消/检测断开后关闭上游。已经执行的工具动作不会被自动撤销，计费也不保证立刻停止。
  • Semaphore 只限制本进程并发，不是用户限流，更不能替代认证。

验收与部署边界 ​

试温度换算、普通问候、超范围温度、超过输入长度、错误 Key、生成中点击停止。中断时不能显示「完成」，半段回答应标注不完整。

本版仅绑定本机，无认证，不可直接作为公开收费 API 部署。上线前补认证、每用户配额、超时、日志脱敏、服务端请求大小限制和部署网关配置。

你的文档站可部署 Cloudflare Pages；这个 Python 服务需要 Python 运行环境。可以将前端放 Pages、后端部署到合适的服务，再配置 HTTPS、确切允许的 CORS 来源和认证。不要尝试把普通 FastAPI 当作纯静态文件上传。

升级练习：加多轮历史（完整事务裁剪）、组件化工具状态、带取消状态的任务列表，再接入项目 2 的知识库检索工具。

用心写给每一个想入门 Agent 开发的前端工程师

© Agent 开发之路