构建一个可用于生产的、工具增强型 LLM Agent，使其具备 Token 流式输出、代码执行、搜索能力，并利用 FastAPI 实现高性能 API 服务。
ChatGPT 的出现带来了震撼的体验，但开发者很快开始思考：如何超越“聊天”本身？我们能否构建一个能够实时推理、联网搜索、执行代码、查询数据，并像人类打字一样流式响应的智能体？
答案是肯定的。通过结合 LangChain 与 FastAPI，我构建了一个多工具 AI Agent。它能够协调调用浏览器、数据库、Python 解释器等外部工具，并以 Token 为单位实时流式生成内容。本文将详细拆解这一实战方案。
🛠️ 愿景：多工具 Agent 是 LLM 能力的延伸
大型语言模型擅长文本生成，但在“执行动作”方面存在局限。LangChain 的 AgentExecutor 为此提供了解决方案——它通过让模型调用各类“工具”来扩展其能力边界。需要实时信息？调用搜索 API。需要计算？启动代码解释器。需要查询数据？连接数据库。
然而，许多演示仅停留在“能运行”的阶段。我们的目标是构建一个能在生产环境稳定运行的系统，充分利用 FastAPI 的异步特性，并实现类似 ChatGPT 的 Token 级流式输出。
🧱 核心技术栈
以下是构建此实时 Agent 的核心技术选型：
- LangChain：负责 Agent 逻辑编排、提示词模板、记忆管理及工具链。
- OpenAI GPT-4 / GPT-3.5：作为核心的推理与生成模型。
- FastAPI：提供高效、异步的 HTTP API 服务。
- AsyncLangChain + SSE：通过 Server-Sent Events 实现实时 Token 流式传输。
- Python REPL Tool：在安全沙箱中执行 Python 代码。
- Google Search Tool：基于 SerpAPI 获取最新网络信息。
- PostgreSQL Tool：查询结构化数据库。
🧠 使用 LangChain 构建 Agent 核心
LangChain 提供了构建可推理、可调用工具的 Agent 所需的核心组件。其基本构建流程如下：
-
定义工具
工具本质上是一个带有名称、描述和执行函数的 Python 包装器。例如，一个简单的代码执行工具：from langchain.agents import Tool
def run_code(code: str) -> str:
# 在沙箱中执行代码（生产环境需谨慎！）
try:
return str(eval(code))
except Exception as e:
return str(e)code_tool = Tool(
name=”Python REPL”,
func=run_code,
description=”Executes Python code and returns the result”
) -
初始化 Agent
使用 LangChain 的initialize_agent()函数，并选择如chat-zero-shot-react-description这类 Agent 类型，使模型能够自主选择工具并解释其推理步骤。 -
添加记忆（可选）
可以使用ConversationBufferMemory来维持会话历史。对于规模化部署，可考虑替换为 Redis 等外部存储。 -
启用流式输出
要实现按 Token 流式输出，需要将AsyncIteratorCallbackHandler这类流式处理器传递给 OpenAI LLM 实例。
🚀 使用 FastAPI 与 Server-Sent Events 提供流式 API
流式传输是提升 LLM 应用用户体验的关键。传统的“请求-等待-完整响应”模式体验不佳。
FastAPI 对 SSE（Server-Sent Events） 的良好支持，使其成为实现低延迟流式传输的理想选择。核心实现方式如下：
from fastapi import FastAPI
from sse_starlette.sse import EventSourceResponse
@app.get("/agent/stream")
async def stream_agent_response(query: str):
async def event_generator():
async for chunk in run_agent_stream(query):
yield {"event": "message", "data": chunk}
return EventSourceResponse(event_generator())
run_agent_stream() 函数会实时产生来自 LangChain Agent 的 Token 片段。前端可以像 OpenAI Playground 一样，逐词渲染响应内容。
🔍 工具集成：搜索、数据库与代码执行
🔎 网络搜索工具
为了让 Agent 能回答实时性问题，通过 LangChain 集成了 SerpAPI。这使得模型在缺乏知识时能够主动搜索网络。
from langchain.tools import Tool
from langchain.utilities import SerpAPIWrapper
search = SerpAPIWrapper()
search_tool = Tool(
name="Google Search",
func=search.run,
description="Use this to look up current events or recent data"
)
📦 PostgreSQL 数据库工具
通过 LangChain 的 SQL Agent 连接 PostgreSQL 数据库，使模型能够查询产品信息、日志、分析数据等结构化内容。这对构建企业内部智能助手尤其有价值。
⚙️ 代码执行工具
为了安全地执行逻辑，我们构建了一个限制全局变量的最小化 Python REPL 沙箱。注意：生产环境中应避免直接使用 eval()，建议采用 Docker 容器隔离或使用专门的代码解释器 API。
🎯 最终效果：实时交互体验
当用户提出一个复杂请求时，例如：
“请总结这篇文章，计算每个段落的平均词数，并附上排名前三的来源链接。”
Agent 可能会执行以下操作：
1. 调用搜索工具查找文章。
2. 使用代码工具处理文本并计算。
3. 查询数据库核实来源信息。
4. 同时，将推理和生成的答案以 Token 流的形式实时推送到前端。
整个过程是“直播”进行的，用户可以看到 Agent 边思考、边行动、边回答，体验更接近与真人协作。
⚡ 关键实践经验
- 流式传输至关重要：Token 级的流式输出能极大提升应用的响应速度和用户信任感。
- LLM 需与工具结合：要超越纯文本生成，必须为其集成外部工具以执行具体任务。
- LangChain 灵活但复杂：需要准备好处理和管理相对复杂的提示词与链式调用。
- FastAPI 异步优势明显：其高性能和原生异步支持非常适合此类实时应用，SSE 集成顺畅。
- 生产环境安全第一：在公开部署前，务必为 API 添加限流、完善的日志记录和监控。
📦 部署方案
可以使用 Docker 将服务容器化，通过 NGINX 作为反向代理，并部署到 Render 或 Fly.io 等云平台。所有敏感信息（如 LLM API 密钥、数据库凭证）应通过环境变量或密钥管理服务安全存储。
前端可以采用 React 等框架，配合用于接收 SSE 的客户端库（如 EventSource 或 SWR），实现几乎无延迟的实时交互界面。
🗣️ 结语
这个项目标志着我们从“提示词工程”迈向了 工具编排 的新阶段。LLM 不再仅仅是内容生成器，而是能够驱动复杂工作流的智能决策中心。借助 LangChain、FastAPI 以及精心挑选的工具集，你可以构建出真正能“行动”的智能体，而不仅仅是会“聊天”的机器人。
关注“鲸栖”小程序，掌握最新AI资讯
本文由鲸栖原创发布，未经许可，请勿转载。转载请注明出处：http://www.itsolotime.com/archives/13070