从零搭建最小可用Python Agent服务：工程化模型调用的完整实践

作者：起个名字好难2026.11.07 03:24浏览量：0

简介：本文面向希望将大模型调用从“临时脚本”升级为“可维护服务”的开发者。你将跟随一个渐进式案例，从项目骨架搭建开始，逐步完成配置管理、数据模型定义、模型客户端封装、服务编排、安全防护与测试反馈等关键环节，最终得到一个具备健康检查、对话、计划生成与文本入库能力的Agent服务骨架。全文强调工程化思维与最小可用原则，助你避开常见陷阱，快速构建出可被其他系统稳定调用的模型服务。

在人工智能应用落地过程中，一个常见现象是：模型调用Demo能跑通，但一旦需要接入真实业务系统，便会暴露出结构混乱、配置硬编码、缺乏安全边界、无法测试等一系列问题。本文将引导你构建一个最小但完整的Python Agent服务，它不仅是一个模型调用接口，更是一个具备清晰分层、可配置、可测试的工程实体。

为什么需要一个“服务”而非“脚本”

一次性的模型调用脚本只能解决“让模型说一句话”的问题。而当我们需要将模型能力嵌入到例如Spring Boot等后端系统中时，就必须考虑以下需求：

  • 可观测性：系统需要知道服务是否存活、加载了何种模型配置。
  • 标准化交互：外部系统需要以统一的数据格式（如JSON）进行通信。
  • 安全与防护：用户输入可能包含恶意内容，必须在进入核心链路前进行过滤。
  • 可扩展性：后续可能要接入检索增强生成、外部工具调用等能力，服务架构需要预留插槽。

因此，我们需要的不是一个函数，而是一个进程。这个进程对外暴露HTTP接口，内部则包含配置层、数据模型层、客户端层、业务编排层和测试反馈层。

项目整体规划与模块划分

本章的目标是搭建一个最小可运行的服务，其核心模块划分如下：

  1. API层：提供/health、/agent/chat、/agent/today、/rag/ingest等端点。
  2. 配置层：集中管理模型参数、API密钥、开关项等。
  3. 数据模型层（DTO）：定义请求与响应的数据结构，确保数据交互的规范性。
  4. 服务编排层：负责处理核心业务逻辑，如输入护栏、检索增强、模型调用等。
  5. RAG存储层：一个极简的向量存储或文档存储实现，用于支撑知识检索。
  6. 安全与测试：包含输入过滤机制以及便于验证的测试命令。

在动手写代码之前，请务必设定一个严格的复制规则：后续所有代码块将分为三类。可执行命令需要你在终端中运行；待写入文件需要完整复制到指定的项目路径中；理解片段仅用于解释原理，不应出现在你的项目中。建议每完成2至4个文件，就运行一次轻量验证，而非等到全部代码写完后才排查错误。

第一步：搭建项目骨架与虚拟环境

首先，我们需要建立一个清晰的目录结构，这比代码本身更重要。清晰的目录可以帮助后续维护者快速定位问题。

  1. # 可执行命令：创建项目根目录
  2. mkdir python-agent-service
  3. cd python-agent-service
  5. # 创建子目录，分别存放API、核心逻辑、数据模型、工具类
  6. mkdir -p app/api app/core app/dto app/infra tests

接下来，初始化一个Python虚拟环境，以避免依赖冲突。

  1. # 可执行命令：创建并激活虚拟环境
  2. python3 -m venv venv
  3. source venv/bin/activate

现在，我们需要安装核心依赖。本服务将使用一个轻量级的Web框架，以及用于环境变量管理的常用库。

  1. # 可执行命令：安装依赖
  2. pip install fastapi uvicorn pydantic python-dotenv

依赖安装完成后，我们可以手动创建一个入口文件，用于启动服务。在这个阶段，我们只需要一个能返回“服务存活”状态的占位服务，以便验证环境是否可用。

  1. # [写入文件] app/main.py
  2. from fastapi import FastAPI
  4. app = FastAPI(title="Python Agent Service")
  6. @app.get("/health")
  7. async def health_check():
  8.     return {"status": "ok"}

为了更优雅地启动服务，我们创建一个启动脚本。

  1. # [写入文件] run.py
  2. import uvicorn
  4. if __name__ == "__main__":
  5.     uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)

执行验证：在终端运行 python run.py。若不报错，则打开浏览器访问 http://localhost:8000/health，应看到 {"status":"ok"}。至此，项目骨架搭建完成。

第二步：配置管理与DTO定义

一个工程化的服务，绝不能将模型名称、API密钥等参数硬编码在业务逻辑中。我们将使用环境变量结合.env文件来管理配置。

  1. # [写入文件] app/core/config.py
  2. from pydantic_settings import BaseSettings
  3. from dotenv import load_dotenv
  5. load_dotenv()
  7. class Settings(BaseSettings):
  8.     # 模型调用开关。生产环境必须显式开启，测试环境默认为False
  9.     model_enabled: bool = False
  10.     # 模型服务地址，可以指向本地代理或云端
  11.     model_base_url: str = "http://localhost:11434"
  12.     # 模型名称
  13.     model_name: str = "agent-smol"
  14.     # 温度参数，控制模型输出的随机性
  15.     model_temperature: float = 0.3
  16.     # API密钥，若使用需要鉴权的云端模型，则填写
  17.     api_key: str = ""
  19. settings = Settings()

接下来，我们定义数据传输对象（DTO）。DTO是API层与业务层之间的契约，它保证了数据结构的稳定性。例如，对话请求需要包含用户消息，同时可能需要携带会话ID用于状态跟踪。

  1. # [写入文件] app/dto/agent.py
  2. from pydantic import BaseModel, Field
  4. class ChatRequest(BaseModel):
  5.     message: str = Field(..., description="用户输入消息")
  6.     session_id: str = Field(default="default", description="会话标识")
  8. class ChatResponse(BaseModel):
  9.     reply: str
  10.     session_id: str
  12. # 定义Today计划请求
  13. class TodayPlanRequest(BaseModel):
  14.     user_profile: dict
  15.     checkin_history: list[dict]

这些模型不仅用于请求校验，也为未来的接口文档生成提供了依据。理解片段：我们在DTO中为每个字段添加了语义描述，在FastAPI中，这些描述会自动出现在交互式文档中，这对于前后端联调非常有益。

第三步：模型客户端与服务编排

这一层是服务的核心。我们需要封装一个统一的大模型客户端，屏蔽底层HTTP请求的差异。同时，服务编排层将负责调用顺序：先执行输入护栏，再调用RAG检索，最后合成提示词发送给模型。

  1. # [写入文件] app/core/model_client.py
  2. import httpx
  3. from typing import Optional
  4. from app.core.config import settings
  6. class ModelClient:
  7.     def __init__(self):
  8.         self.base_url = settings.model_base_url
  9.         self.api_key = settings.api_key
  11.     def chat(self, messages: list[dict], temperature: float = None) -> str:
  12.         # 构建请求数据体，使用行业标准的chat/completions接口
  13.         payload = {
  14.             "model": settings.model_name,
  15.             "messages": messages,
  16.             "temperature": temperature or settings.model_temperature,
  17.             "stream": False,
  18.         }
  19.         headers = {}
  20.         if self.api_key:
  21.             headers["Authorization"] = f"Bearer {self.api_key}"
  22.         # 这里使用httpx通用请求，具体URL由配置决定
  23.         response = httpx.post(f"{self.base_url}/v1/chat/completions", json=payload, headers=headers)
  24.         response.raise_for_status()
  25.         return response.json()["choices"][0]["message"]["content"]

在服务编排层，我们实现一个AgentService类，将护栏、RAG和模型调用整合起来。注意，这里要体现“默认测试不需要真实密钥”的逻辑——当model_enabled为False时，我们应返回模拟数据或固定回复，便于离线开发。

  1. # [写入文件] app/core/agent_service.py
  2. from app.dto.agent import ChatRequest, ChatResponse
  3. from app.core.model_client import ModelClient
  4. from app.core.config import settings
  6. class AgentService:
  7.     def __init__(self):
  8.         self.client = ModelClient()
  10.     def _guardrail(self, message: str) -> bool:
  11.         # 简单护栏：检查消息是否包含异常内容，例如过长或包含特殊字符
  12.         if len(message) > 500:
  13.             return False
  14.         # 此处可扩展敏感词过滤逻辑
  15.         return True
  17.     def chat(self, req: ChatRequest) -> ChatResponse:
  18.         if not self._guardrail(req.message):
  19.             return ChatResponse(reply="抱歉，您输入的文本长度超出限制或包含不合适的内容。", session_id=req.session_id)
  21.         # 若未开启模型，则返回模拟回复
  22.         if not settings.model_enabled:
  23.             reply = f"[模拟回复] 您说的是：{req.message}"
  24.             return ChatResponse(reply=reply, session_id=req.session_id)
  26.         # 真实模型链路（需显式开启）
  27.         messages = [{"role": "system", "content": "你是一个实用的Agent助手。"}, {"role": "user", "content": req.message}]
  28.         reply = self.client.chat(messages)
  29.         return ChatResponse(reply=reply, session_id=req.session_id)
  31. agent_service = AgentService()

理解片段：注意model_enabled开关的作用。在开发阶段，我们无需任何外部依赖即可测试HTTP接口和业务逻辑。当需要验证真实模型效果时，再将此开关设为True并配置正确的模型地址。这种设计保证了服务在无密钥环境下依然可被集成测试。

第四步：实现最小RAG存储与接入

检索增强生成（RAG）是当下Agent服务的核心能力之一。本章我们实现一个最简版本：将文档切块后存入内存列表，并在调用模型前检索相关片段。

  1. # [写入文件] app/infra/rag_store.py
  2. class SimpleRagStore:
  3.     def __init__(self):
  4.         self.chunks = []
  6.     def ingest(self, text: str, chunk_size: int = 100):
  7.         # 简单切块：按固定大小切分，不重叠
  8.         for i in range(0, len(text), chunk_size):
  9.             self.chunks.append(text[i:i+chunk_size])
  11.     def search(self, query: str, top_k: int = 2) -> list[str]:
  12.         # 最朴素的检索：基于关键词匹配
  13.         words = query.split()
  14.         scored = []
  15.         for chunk in self.chunks:
  16.             score = sum(1 for word in words if word in chunk)
  17.             scored.append((score, chunk))
  18.         scored.sort(reverse=True, key=lambda x: x[0])
  19.         return [chunk for _, chunk in scored[:top_k]]
  21. rag_store = SimpleRagStore()

接下来，我们需要在HTTP层暴露一个接口，用于将文本写入RAG存储。

  1. # [写入文件] app/api/routes.py
  2. from fastapi import APIRouter
  3. from app.dto.agent import ChatRequest, ChatResponse
  4. from app.core.agent_service import agent_service
  5. from app.infra.rag_store import rag_store
  6. from pydantic import BaseModel
  8. router = APIRouter()
  10. class IngestRequest(BaseModel):
  11.     text: str
  13. @router.post("/agent/chat", response_model=ChatResponse)
  14. async def chat(req: ChatRequest):
  15.     return agent_service.chat(req)
  17. @router.post("/rag/ingest")
  18. async def ingest(req: IngestRequest):
  19.     rag_store.ingest(req.text)
  20.     return {"status": "ok", "chunks": len(rag_store.chunks)}
  22. @router.get("/health")
  23. async def health():
  24.     return {"status": "ok", "model_config": {"enabled": settings.model_enabled, "name": settings.model_name}}

在main.py中注册该路由，使接口生效。

  1. # [覆盖写入] app/main.py
  2. from fastapi import FastAPI
  3. from app.api.routes import router
  5. app = FastAPI(title="Python Agent Service")
  6. app.include_router(router)

执行验证：重启服务后，使用curl命令测试接口：

  1. # 可执行命令：测试服务
  2. curl -X POST http://localhost:8000/rag/ingest -H "Content-Type: application/json" -d '{"text":"百度智能云提供丰富的人工智能服务。"}'
  3. curl -X POST http://localhost:8000/agent/chat -H "Content-Type: application/json" -d '{"message":"什么是百度智能云？"}'

若返回了包含模拟回复的JSON，则说明RAG接入基本贯通。

第五步：实现“今日计划”生成功能

最后，我们提供一个业务示例接口，展示如何将用户档案与打卡记录结合，生成个性化内容。该接口的核心在于提示词工程——如何将结构化数据转变为模型可理解的指令。

  1. # [写入文件] app/core/today_service.py
  2. from app.dto.agent import TodayPlanRequest
  4. def generate_today_plan(req: TodayPlanRequest) -> str:
  5.     profile = req.user_profile
  6.     history = req.checkin_history
  7.     # 在真实实现中，这里会调用模型客户端
  8.     if not settings.model_enabled:
  9.         plan = f"根据档案{profile}和最近{len(history)}天的记录，建议今日先完成最重要的一项任务，并抽出30分钟进行复盘。"
  10.         return plan
  11.     # 真实模型调用链路（省略prompt构建细节）
  12.     return ""

在路由中注册该服务，完成所有接口的搭建。至此，一个具备最小功能的Agent服务已完整落地。

测试与反馈闭环

每完成一个模块，都应进行轻量验证。我们建议在项目根目录创建一个测试脚本，检查核心接口是否返回预期结果。

  1. # [执行命令] 运行基础测试
  2. python -c "from app.core.agent_service import agent_service; print(agent_service.chat(ChatRequest(message='hello')))"

工程化开发的核心在于“小步快跑”。通过频繁验证，可以避免错误堆积，确保每个阶段都有可靠的产出。

总结与展望

本章完成的服务虽然代码量不大，但它具备了一个生产级Agent服务的基本要素：清晰的配置层、严格的数据模型、可切换的模型客户端、基础安全防护以及可扩展的RAG骨架。下一步，你可以在此基础上增加真实的向量数据库、流式输出、多轮对话管理等功能，逐步将其锤炼为一个高性能的智能体服务。记住，工程化的起点不是复杂的架构设计，而是对每一个细节的严谨态度。

最热文章