从零构建可被后端调用的 Python Agent 服务：最小工程化实践

作者：暴富20212026.11.06 20:11浏览量：0

简介：本文不追求“让模型说一句话”，而是演示如何把一个 Python Agent 能力封装成可被后端服务调用的最小工程结构。你将看到配置管理、DTO 设计、客户端隔离、服务编排、安全护栏、轻量 RAG 与测试反馈如何同时落地，并通过分阶段验证降低调试成本。

一、为什么是“工程化”而不是“一次调用”

很多开发者第一次接触智能体开发时，习惯在脚本里直接调用模型接口，输入一句话，拿到一句回复，然后认为“已经会了”。这种方式适合验证模型可用性，但距离真正可被业务系统消费的 Agent 服务还有很大差距。

一个可被后端调用的 Agent 服务至少需要回答几个问题：服务如何启动和停止？如何暴露健康检查？如何接收结构化请求并返回一致响应？模型配置从哪里读取？当用户输入存在风险时如何处理？知识检索和工具调用如何进入主链路？没有真实模型密钥时如何测试？

本文的目标就是把这些工程问题压缩到一个最小但完整的 Python 服务中。服务本身不依赖特定模型厂商，所有模型调用通过一层客户端抽象完成；默认测试使用假客户端，真实模型测试必须显式开启。这样既保证学习过程可复现，也避免误调线上接口。

二、整体结构设计

整个服务围绕一个轻量 HTTP 框架组织，但核心逻辑不绑定框架细节。目录结构可以这样设计：

  1. agent-service/
  2. ├── app/
  3. │   ├── __init__.py
  4. │   ├── config.py
  5. │   ├── dto.py
  6. │   ├── clients/
  7. │   │   ├── __init__.py
  8. │   │   ├── base.py
  9. │   │   ├── fake.py
  10. │   │   └── real.py
  11. │   ├── services/
  12. │   │   ├── __init__.py
  13. │   │   ├── guard.py
  14. │   │   ├── rag.py
  15. │   │   ├── agent.py
  16. │   │   └── today.py
  17. │   └── routes.py
  18. ├── tests/
  19. │   ├── test_health.py
  20. │   ├── test_chat.py
  21. │   └── test_rag.py
  22. ├── requirements.txt
  23. └── README.md

这个结构并不复杂，但它把关注点分离得足够清楚：配置只存在于 config.py；请求和响应结构集中在 dto.py；所有外部模型访问都通过 clients 层隔离；业务编排放在 services 层；路由只负责协议适配。

三、配置管理：让服务可运行、可观测、可切换

配置管理容易被忽视。如果模型地址、密钥、开关散落在代码各处，后续维护会非常痛苦。我们采用集中式配置对象，并使用环境变量覆盖默认值。

  1. import os
  2. from dataclasses import dataclass
  4. @dataclass(frozen=True)
  5. class Settings:
  6.     app_name: str = "minimal-agent-service"
  7.     model_provider: str = "fake"
  8.     model_name: str = "fake-model"
  9.     api_key: str = ""
  10.     base_url: str = ""
  11.     use_real_model: bool = False
  12.     rag_chunk_size: int = 500
  13.     rag_top_k: int = 3
  15.     @classmethod
  16.     def from_env(cls) -> "Settings":
  17.         return cls(
  18.             model_provider=os.getenv("MODEL_PROVIDER", "fake"),
  19.             model_name=os.getenv("MODEL_NAME", "fake-model"),
  20.             api_key=os.getenv("API_KEY", ""),
  21.             base_url=os.getenv("BASE_URL", ""),
  22.             use_real_model=os.getenv("USE_REAL_MODEL", "false").lower() == "true",
  23.             rag_chunk_size=int(os.getenv("RAG_CHUNK_SIZE", "500")),
  24.             rag_top_k=int(os.getenv("RAG_TOP_K", "3")),
  25.         )

关键点是 use_real_model 必须显式设置为 true 才会启用真实调用。默认情况下，服务运行在“安全模式”，所有模型相关测试不会产生外部请求。这在开发早期尤其重要，可以避免因为配置错误导致的意外调用。

四、DTO 设计：统一请求与响应

HTTP 层适合传递 JSON，但服务内部如果直接操作字典，会很快失去类型约束。我们定义简单的数据类来描述请求和响应。

  1. from dataclasses import dataclass, field
  2. from typing import Any, Dict, List, Optional
  4. @dataclass
  5. class ChatRequest:
  6.     message: str
  7.     user_id: Optional[str] = None
  8.     session_id: Optional[str] = None
  10. @dataclass
  11. class ChatResponse:
  12.     reply: str
  13.     guarded: bool = False
  14.     retrieved_chunks: List[Dict[str, Any]] = field(default_factory=list)
  15.     tool_calls: List[Dict[str, Any]] = field(default_factory=list)
  16.     model: str = ""
  18. @dataclass
  19. class HealthResponse:
  20.     status: str
  21.     model_provider: str
  22.     model_name: str
  23.     use_real_model: bool

这样做的好处是：路由层只负责把 JSON 转换成 DTO；服务层只处理 DTO；测试中构造对象也更直观。后续如果增加字段，只需在数据类中修改，影响面可控。

五、模型客户端隔离：假的先跑通，真的再替换

模型客户端是服务中最容易产生外部依赖的部分。我们定义一个极简抽象，然后提供两个实现：假客户端用于测试，真客户端用于真实调用。抽象可以只保留一个 chat 方法。

  1. from typing import Any, Dict, List
  3. class BaseModelClient:
  4.     def chat(
  5.         self,
  6.         messages: List[Dict[str, str]],
  7.         context: str = "",
  8.         tools: List[Dict[str, Any]] = None,
  9.     ) -> Dict[str, Any]:
  10.         raise NotImplementedError

假客户端返回确定性内容，方便断言：

  1. class FakeModelClient(BaseModelClient):
  2.     def chat(self, messages, context="", tools=None):
  3.         return {
  4.             "content": f"fake reply for: {messages[-1]['content']}",
  5.             "tool_calls": [],
  6.         }

真客户端则通过通用 HTTP 方式调用模型服务，具体地址、密钥和模型名从配置读取。实现时尽量避免绑定某个厂商的 SDK，只使用标准 HTTP 库和通用消息格式。这样当模型提供方变化时，只需调整配置和适配逻辑，不影响上层编排。

六、输入护栏：在进入主链路之前拦截风险

用户消息不能直接进入模型。最小可行做法是在服务编排前设置输入护栏，处理明显不适合处理的内容：例如超长输入、空消息、包含明确越权指令或敏感操作的请求。

  1. class InputGuard:
  2.     def __init__(self, max_length: int = 2000):
  3.         self.max_length = max_length
  5.     def check(self, message: str) -> tuple[bool, str]:
  6.         if not message or not message.strip():
  7.             return False, "empty message"
  8.         if len(message) > self.max_length:
  9.             return False, "message too long"
  10.         lowered = message.lower()
  11.         blocked_phrases = ["ignore previous instructions", "system prompt"]
  12.         for phrase in blocked_phrases:
  13.             if phrase in lowered:
  14.                 return False, "prompt injection detected"
  15.         return True, "ok"

护栏不是安全审计的全部，但它提供了一个明确的位置来集中管理基础规则。后续可以在这里增加更复杂的检测逻辑，例如敏感信息识别或异常模式匹配，而不必修改主业务流程。

七、RAG 与工具链路：让回复有上下文

即使是最小版本，Agent 也不应只依赖模型参数记忆。我们引入一个极简 RAG 存储：将文本按块切分，存储到内存中，并在对话时做相似度检索。这里不依赖外部向量数据库，只使用通用文本相似度计算。

  1. import re
  2. from typing import List, Dict
  4. class MinimalRagStore:
  5.     def __init__(self, chunk_size: int = 500):
  6.         self.chunk_size = chunk_size
  7.         self.chunks: List[str] = []
  9.     def ingest(self, text: str) -> int:
  10.         normalized = re.sub(r"\s+", " ", text).strip()
  11.         chunks = [
  12.             normalized[i:i + self.chunk_size]
  13.             for i in range(0, len(normalized), self.chunk_size)
  14.         ]
  15.         self.chunks.extend(chunks)
  16.         return len(chunks)
  18.     def search(self, query: str, top_k: int = 3) -> List[Dict[str, any]]:
  19.         if not self.chunks:
  20.             return []
  21.         query_terms = set(query.lower().split())
  22.         scored = []
  23.         for idx, chunk in enumerate(self.chunks):
  24.             chunk_terms = set(chunk.lower().split())
  25.             score = len(query_terms & chunk_terms)
  26.             scored.append((idx, chunk, score))
  27.         scored.sort(key=lambda x: x[2], reverse=True)
  28.         return [
  29.             {"index": idx, "text": chunk, "score": score}
  30.             for idx, chunk, score in scored[:top_k] if score > 0
  31.         ]

检索结果会作为上下文拼接到主模型请求中。这样做有两个目的：一是验证 RAG 链路是否通顺；二是让后续替换更复杂的检索方案时，接口保持不变。

工具调用在最小版本中可以先实现为一个固定工具注册表，例如“获取今日日期”“查询用户打卡状态”。由于默认使用假客户端，工具调用不会真实执行；但在真客户端模式下，工具定义会随请求一起传递，并通过返回的调用信息在服务层执行对应函数。

八、服务编排：把护栏、RAG、工具和主模型串起来

AgentService 是核心编排层。它会依次完成：检查输入、检索上下文、决定是否调用工具、准备消息、调用模型客户端、组装统一响应。

  1. class AgentService:
  2.     def __init__(self, client, guard, rag_store, tools=None):
  3.         self.client = client
  4.         self.guard = guard
  5.         self.rag_store = rag_store
  6.         self.tools = tools or []
  8.     def chat(self, message: str) -> ChatResponse:
  9.         allowed, reason = self.guard.check(message)
  10.         if not allowed:
  11.             return ChatResponse(reply=f"输入被拦截: {reason}", guarded=True)
  13.         chunks = self.rag_store.search(message)
  14.         context = "\n".join([c["text"] for c in chunks])
  16.         messages = [
  17.             {"role": "system", "content": "你是一个受控助手。"},
  18.             {"role": "user", "content": message},
  19.         ]
  21.         result = self.client.chat(
  22.             messages=messages,
  23.             context=context,
  24.             tools=self.tools,
  25.         )
  27.         return ChatResponse(
  28.             reply=result.get("content", ""),
  29.             retrieved_chunks=chunks,
  30.             tool_calls=result.get("tool_calls", []),
  31.             model=getattr(self.client, "model_name", ""),
  32.         )

这里的关键不是逻辑多么复杂，而是每一段职责都很清楚。后续如果需要增加多轮记忆、异步处理或流式返回，都可以在对应位置扩展，而不需要重写整个服务。

九、路由与健康检查：让服务成为服务

HTTP 路由层把上述能力暴露为接口。/health 返回当前配置和模型模式，方便部署时做探活；/agent/chat 接收用户消息并返回 ChatResponse；/agent/today 基于用户档案和打卡记录生成今日计划；/rag/ingest 接收文本并写入 RAG 存储。

  1. def register_routes(app, settings, agent_service, today_service, rag_store):
  2.     @app.get("/health")
  3.     def health():
  4.         return {
  5.             "status": "ok",
  6.             "model_provider": settings.model_provider,
  7.             "model_name": settings.model_name,
  8.             "use_real_model": settings.use_real_model,
  9.         }
  11.     @app.post("/agent/chat")
  12.     def chat(req: ChatRequest):
  13.         return agent_service.chat(req.message)
  15.     @app.post("/agent/today")
  16.     def today(req: TodayRequest):
  17.         return today_service.build_plan(req.user_id)
  19.     @app.post("/rag/ingest")
  20.     def ingest(req: IngestRequest):
  21.         count = rag_store.ingest(req.text)
  22.         return {"ingested_chunks": count}

/health 的价值不只是返回 200。它把模型模式和配置暴露出来，使运维人员可以快速确认当前环境是安全模式还是真实模式，避免在错误环境中排查问题。

十、测试策略：每两到四个文件就验证一次

本文强调分阶段验证。每完成两到四个文件，就运行一次轻量测试，而不是等全部代码写完再调试。以下是一个典型的阶段划分：

第一阶段：配置与 DTO。验证配置可以从环境变量正确读取，DTO 可以正常实例化。

第二阶段：假客户端与输入护栏。验证假客户端返回确定性内容，护栏能拦截空消息和超长消息。

第三阶段：RAG 存储。写入一段文本，确认切片数量正确，检索能返回相关块。

第四阶段：服务编排与路由。使用测试客户端发送请求，确认 /health、/agent/chat、/rag/ingest 均正常。

这种做法的核心收益是：错误一旦出现，只可能出现在最近两到四个文件之间，范围很小，定位很快。

十一、真实模型切换的工作量

启用真实模型时，不需要修改服务编排和路由，只需要在环境变量中显式设置：

  1. export USE_REAL_MODEL=true
  2. export MODEL_PROVIDER=real
  3. export MODEL_NAME=你的模型名
  4. export API_KEY=你的密钥
  5. export BASE_URL=模型服务地址

然后运行服务。此时 clients.real 会代替 clients.fake，并由真实模型完成回复。由于所有测试默认使用假客户端，在开发环境中可以随时运行测试套件而不产生外部调用。

十二、这个最小版本到底能带来什么

完成这个结构之后，你得到的不是一个“会说话的脚本”，而是一个可以持续演进的 Agent 服务骨架。它解决了几个实际问题：第一，模型能力被封装在明确边界内，替换成本低；第二，输入风险和运行模式有基本控制；第三，RAG 和工具链路不是概念，而是真实可调的代码；第四，测试可以在无外部依赖的情况下运行；第五，后端服务可以通过 HTTP 接口直接集成。

后续如果要继续演进，可以围绕这个骨架逐步增加多轮会话、持久化检索、更完善的工具执行器、异步队列、流式输出和可观测性。每一步都能在现有结构中自然生长，而不是推倒重来。

这就是“最小可工程化 Agent 服务”的意义：它足够小，能在一个章节中讲完；也足够完整，能支撑后续的严肃开发。

最热文章