大模型应用开发实战（18）——构建智能体（Agent）框架客户端

🤵‍♂️ 个人主页：小李同学_LSH的主页

✍🏻 作者简介：LLM学习者
🐋 希望大家多多支持，我们一起进步！😄
如果文章对你有帮助的话，
欢迎评论 💬点赞👍🏻 收藏 📂加关注+

目录

一、先把目标说清楚：要做什么

二、为什么 Agent 一旦上服务端，问题就变了

三、项目结构

四、先把基础 LLM 调用封装好，这是服务端的第一块地基

config.py

.env

core/llm.py

core/message.py

五、代码实践

很多人做到 Agent 这一步，前面的路都还算顺：
封装一个 LLM 类，补几个工具，写个 ReAct 循环，命令行里一跑，智能体就“活了”。

但问题也恰恰从这里开始。

一旦你不再满足于本地 Demo，而是想把它真正变成一个可访问、可复用、可扩展的服务端系统，原来那套写法就会迅速暴露问题：

• 对话历史只存在进程内，服务一重启就没了
• 多用户同时访问时，上下文容易串线
• 工具调用、异常处理、接口返回格式都不统一
• 调试靠print，排障靠猜
• 本地能跑，一上 HTTP 服务就开始混乱

所以，从 Agent 到 Agent Server，并不是“把脚本套个 FastAPI”这么简单。
真正要做的，是把原来“面向单次运行”的代码，重构成“面向请求生命周期”的服务架构。

Hello-Agents 在第四章里对基础 LLM 调用做了一个很关键的动作：先把模型 ID、API Key、Base URL 统一放进环境变量，再封装一个专门的HelloAgentsLLM客户端类，把与模型服务交互的细节和上层 Agent 逻辑分开。这个思路非常适合作为服务端化的起点。

这篇文章就不空谈概念，直接从工程视角出发，带你把一个最小可用的 Agent 框架，继续往前推成一个服务端版本。

一、先把目标说清楚：要做什么

我们最终要得到的，不再是这样一个命令行 Agent：

result = agent.run("请帮我总结这段文本") print(result)

而是一个可以通过 HTTP 调用的服务：

POST /chat POST /chat/stream GET /sessions/{session_id} POST /sessions/{session_id}/clear GET /health

也就是说，我们希望把原本的单机 Agent，升级成下面这类结构：

不是“多了 FastAPI”，而是多了三件关键事情：

1. 请求生命周期
2. 会话状态管理
3. 服务化接口抽象

FastAPI 本身就是一个基于 Python 类型提示的高性能 API 框架，支持自动生成 OpenAPI 文档；它的中间件机制可以在每个请求进入路由前和响应返回前统一处理逻辑，这一点非常适合做 Agent 服务端。

二、为什么 Agent 一旦上服务端，问题就变了

在本地脚本里，Agent 的核心问题通常是：

• 模型怎么调
• Prompt 怎么写
• 工具怎么接
• 循环什么时候停

但服务端场景下，问题会变成：

• 一个用户的多轮消息放哪
• 多个用户同时访问怎么隔离
• 接口怎么定义才不乱
• 流式输出怎么返回
• 模型异常、工具异常、参数异常怎么统一处理
• 服务启动时怎么加载配置
• 健康检查、日志、追踪怎么补

这两个阶段的重心完全不同。

为了更直观一点，可以先看这个对比表。

三、项目结构

agent_server/ ├── app/ │ ├── main.py │ ├── config.py │ ├── schemas.py │ ├── middleware.py │ ├── exceptions.py │ │ │ ├── core/ │ │ ├── llm.py │ │ ├── message.py │ │ └── agent.py │ │ │ ├── services/ │ │ ├── agent_service.py │ │ └── session_service.py │ │ │ ├── tools/ │ │ ├── base.py │ │ ├── registry.py │ │ └── calculator.py │ │ │ └── api/ │ └── chat.py │ ├── .env ├── requirements.txt └── run.py

这个结构对应的是很典型的三层思路：

• core：模型调用、消息对象、Agent 主循环
• services：服务编排、会话管理
• api：HTTP 路由层

四、先把基础 LLM 调用封装好，这是服务端的第一块地基

先用.env管模型 ID、API Key、Base URL，再封装一个统一的 LLM 客户端，让业务逻辑不直接碰底层 SDK。

这个动作看起来普通，但服务端一旦起来，它的价值会立刻放大：

• 不同环境切模型更方便
• 测试环境、生产环境可切换
• 更容易替换成 OpenAI 兼容服务
• 不会把 SDK 调用散落在各个路由里

config.py

from pydantic_settings import BaseSettings class Settings(BaseSettings): app_name: str = "Agent Server" debug: bool = True llm_model: str = "gpt-4.1-mini" llm_api_key: str llm_base_url: str llm_timeout: int = 60 llm_temperature: float = 0.2 max_steps: int = 5 max_history_length: int = 20 class Config: env_file = ".env" settings = Settings()

.env

LLM_API_KEY=your_api_key LLM_BASE_URL=https://your-openai-compatible-endpoint/v1 LLM_MODEL=gpt-4.1-mini

core/llm.py

from openai import OpenAI from app.config import settings class HelloAgentsLLM: def __init__(self): self.model = settings.llm_model self.client = OpenAI( api_key=settings.llm_api_key, base_url=settings.llm_base_url, timeout=settings.llm_timeout, ) def chat(self, messages, tools=None): payload = { "model": self.model, "messages": [m.to_dict() for m in messages], "temperature": settings.llm_temperature, } if tools: payload["tools"] = tools payload["tool_choice"] = "auto" return self.client.chat.completions.create(**payload)

core/message.py

from dataclasses import dataclass, field from typing import Optional, Dict, Any from datetime import datetime @dataclass class Message: role: str content: str name: Optional[str] = None metadata: Dict[str, Any] = field(default_factory=dict) timestamp: datetime = field(default_factory=datetime.utcnow) def to_dict(self): data = { "role": self.role, "content": self.content, } if self.name: data["name"] = self.name return data

五、代码实践

import os from openai import OpenAI from dotenv import load_dotenv from typing import List, Dict # 加载 .env 文件中的环境变量 load_dotenv() class HelloAgentsLLM: """ 为本书 "Hello Agents" 定制的LLM客户端。 它用于调用任何兼容OpenAI接口的服务，并默认使用流式响应。 """ def __init__(self, model: str = None, apiKey: str = None, baseUrl: str = None, timeout: int = None): """ 初始化客户端。优先使用传入参数，如果未提供，则从环境变量加载。 """ self.model = model or os.getenv("LLM_MODEL_ID") apiKey = apiKey or os.getenv("LLM_API_KEY") baseUrl = baseUrl or os.getenv("LLM_BASE_URL") timeout = timeout or int(os.getenv("LLM_TIMEOUT", 60)) if not all([self.model, apiKey, baseUrl]): raise ValueError("模型ID、API密钥和服务地址必须被提供或在.env文件中定义。") self.client = OpenAI(api_key=apiKey, base_url=baseUrl, timeout=timeout) def think(self, messages: List[Dict[str, str]], temperature: float = 0) -> str: """ 调用大语言模型进行思考，并返回其响应。 """ print(f"🧠 正在调用 {self.model} 模型...") try: response = self.client.chat.completions.create( model=self.model, messages=messages, temperature=temperature, stream=True, ) # 处理流式响应 print("✅ 大语言模型响应成功:") collected_content = [] for chunk in response: content = chunk.choices[0].delta.content or "" print(content, end="", flush=True) collected_content.append(content) print() # 在流式输出结束后换行 return "".join(collected_content) except Exception as e: print(f"❌ 调用LLM API时发生错误: {e}") return None # --- 客户端使用示例 --- if __name__ == '__main__': try: llmClient = HelloAgentsLLM() exampleMessages = [ {"role": "system", "content": "You are a helpful assistant that writes Python code."}, {"role": "user", "content": "写一个快速排序算法"} ] print("--- 调用LLM ---") responseText = llmClient.think(exampleMessages) if responseText: print("\n\n--- 完整模型响应 ---") print(responseText) except ValueError as e: print(e) >>> --- 调用LLM --- 🧠 正在调用 xxxxxx 模型... ✅ 大语言模型响应成功: 快速排序是一种非常高效的排序算法...