从零搭建一个可运行的 Python Agent 服务：最小模型调用与工程化实践

作者：很菜不狗2026.11.06 20:00浏览量：0

简介：本文将带你从零构建一个可独立运行的 Python Agent 服务，而非一次性模型调用脚本。你将获得一个具备健康检查、输入护栏、RAG 链路、工具编排与测试开关的工程化服务骨架，可直接被后端系统通过 HTTP 调用。文章重点不在于“让模型说一句话”，而在于把模型调用放进配置、DTO、客户端、服务编排、安全边界与测试反馈共存的工程结构中，为后续接入真实模型与业务链路打下可靠基础。

一、本章目标：从“调通模型”到“跑起服务”

很多开发者在第一次接触大模型应用开发时，习惯性地写一段脚本，把 Prompt 丢给模型，再把返回结果打印出来。这种方式用来验证模型可用性没有问题，但它距离一个真正可以被业务系统集成的 Agent 服务还有明显距离。

本章要完成的事情，不是再写一段“调用—打印—结束”的脚本，而是搭建一个可以独立运行的 Python Agent Service。它具备以下能力：

  • /health 接口可以返回服务运行状态和当前模型配置；
  • /agent/chat 接口可以接收用户消息，先执行输入护栏，再进入 RAG、工具调用和主模型链路；
  • /agent/today 接口可以基于用户档案与打卡记录生成当日计划；
  • /rag/ingest 接口可以将文本切块并写入最小 RAG 存储；
  • 默认测试不依赖真实模型密钥，真实模型测试必须显式开启。

这个服务后续可以被 Spring Boot 等后端系统通过 HTTP 协议直接调用。换句话说，它不是一个“模型 Demo”，而是一个具备基本工程形态的 Agent 服务骨架。

二、为什么“工程结构”比“模型调用”更重要

在真实项目中，模型调用只占整个系统的一小部分。更多的工作量分布在配置管理、数据传输对象设计、客户端封装、服务编排、异常处理、安全边界和测试验证等环节。如果这些环节缺失，模型能力再强，也很难稳定地融入业务系统。

本章的核心关注点可以概括为六个关键词：

  1. 配置：模型地址、密钥、温度、最大 Token 数等参数应当集中管理，而不是散落在代码各处。
  2. DTO：请求与响应结构需要明确定义，避免在业务链路中传递裸字典。
  3. 客户端：模型调用逻辑应封装为独立客户端，便于替换底层模型或接入其他推理服务。
  4. 服务编排：输入护栏、RAG 检索、工具调用、主模型生成等步骤需要有清晰的执行顺序。
  5. 安全边界：用户输入在进入主链路前必须经过校验与过滤。
  6. 测试反馈：每完成若干文件就运行一次轻量验证，避免错误在后期集中爆发。

只有把这些模块同时放进项目里，才算真正把模型调用“工程化”了。

三、代码块分类与复制规则

为了让整个搭建过程清晰可控，本章会涉及三类代码块，必须严格区分：

1. 执行命令

这类代码块用于在终端中运行，例如安装依赖、启动服务、执行测试等。请直接复制到终端执行。

  1. # 示例：安装项目依赖
  2. pip install -r requirements.txt

2. 写入文件

这类代码块需要把内容完整复制到指定文件中。文件名会在代码块上方或内部明确标注。

  1. # 示例：app/config.py
  2. class ModelConfig:
  3.     provider: str = "mock"
  4.     api_key: str = ""
  5.     base_url: str = ""
  6.     temperature: float = 0.7
  7.     max_tokens: int = 2048

3. 理解片段，不要复制

这类代码块仅用于解释某个设计决策或执行流程，不需要写入项目。它们通常是对应模块的简化示意，帮助理解代码为什么“长成这样”。

  1. # 理解片段：请求处理流程示意
  2. 用户消息 -> 输入护栏 -> RAG 检索 -> 工具调用 -> 主模型生成 -> 响应 DTO

每完成 2 到 4 个文件，就应该运行一次轻量验证。不要等整章结束才测试，否则错误会堆积在一起，定位成本会显著上升。

四、项目目录结构设计

一个清晰的项目结构是后续所有工作的基础。推荐采用以下目录组织方式：

  1. agent-service/
  2. ├── app/
  3. │   ├── __init__.py
  4. │   ├── main.py                 # FastAPI 应用入口
  5. │   ├── config.py               # 配置管理
  6. │   ├── dto/
  7. │   │   ├── __init__.py
  8. │   │   ├── chat.py             # 聊天请求/响应 DTO
  9. │   │   ├── today.py            # 今日计划 DTO
  10. │   │   └── ingest.py           # RAG 写入 DTO
  11. │   ├── client/
  12. │   │   ├── __init__.py
  13. │   │   ├── base.py             # 模型客户端抽象
  14. │   │   └── mock_client.py      # 模拟模型客户端
  15. │   ├── guard/
  16. │   │   ├── __init__.py
  17. │   │   └── input_guard.py      # 输入护栏
  18. │   ├── rag/
  19. │   │   ├── __init__.py
  20. │   │   ├── store.py            # 最小 RAG 存储
  21. │   │   └── ingest.py           # 文本切块与写入
  22. │   ├── tool/
  23. │   │   ├── __init__.py
  24. │   │   └── registry.py         # 工具注册表
  25. │   └── service/
  26. │       ├── __init__.py
  27. │       ├── chat_service.py     # 聊天服务编排
  28. │       └── today_service.py    # 今日计划服务编排
  29. ├── tests/
  30. │   ├── __init__.py
  31. │   ├── test_health.py
  32. │   ├── test_chat.py
  33. │   └── test_rag.py
  34. ├── requirements.txt
  35. └── README.md

这个结构遵循“关注点分离”原则：配置、传输对象、模型客户端、服务编排、安全护栏和 RAG 存储各自独立，后续扩展时不会互相干扰。

五、配置模块：集中管理模型参数

配置模块负责加载和管理所有与模型相关的参数。它需要区分“模拟模式”和“真实模式”，默认使用模拟模式，避免在开发和测试阶段误调用真实模型接口。

  1. # app/config.py
  2. from dataclasses import dataclass, field
  3. from enum import Enum
  4. import os
  7. class ModelMode(str, Enum):
  8.     MOCK = "mock"
  9.     REAL = "real"
  12. @dataclass
  13. class ModelConfig:
  14.     mode: ModelMode = ModelMode.MOCK
  15.     api_key: str = ""
  16.     base_url: str = ""
  17.     model_name: str = "default-model"
  18.     temperature: float = 0.7
  19.     max_tokens: int = 2048
  22. @dataclass
  23. class AppConfig:
  24.     model: ModelConfig = field(default_factory=ModelConfig)
  25.     rag_store_path: str = "./data/rag_store.json"
  28. def load_config() -> AppConfig:
  29.     config = AppConfig()
  30.     # 真实模型测试必须显式开启
  31.     mode = os.getenv("AGENT_MODEL_MODE", ModelMode.MOCK.value)
  32.     config.model.mode = ModelMode(mode)
  33.     config.model.api_key = os.getenv("AGENT_MODEL_API_KEY", "")
  34.     config.model.base_url = os.getenv("AGENT_MODEL_BASE_URL", "")
  35.     config.model.model_name = os.getenv("AGENT_MODEL_NAME", "default-model")
  36.     return config

这里有几个设计要点值得注意：

  • 使用环境变量注入敏感信息，避免密钥硬编码；
  • 默认使用模拟模式，降低开发门槛；
  • 真实模式的开启必须显式设置环境变量，防止误调用；
  • 数据类（dataclass）让配置结构一目了然。

六、DTO 模块：定义清晰的请求与响应边界

在服务化场景中，HTTP 接口的请求体和响应体应当有明确的类型定义。这不仅能提升可读性，也能帮助接口文档自动生成。

  1. # app/dto/chat.py
  2. from pydantic import BaseModel, Field
  3. from typing import List, Optional
  6. class ChatRequest(BaseModel):
  7.     message: str = Field(..., min_length=1, max_length=2000, description="用户输入消息")
  8.     session_id: Optional[str] = Field(None, description="会话标识，可选")
  9.     user_profile: Optional[dict] = Field(None, description="用户档案，可选")
  12. class ChatResponse(BaseModel):
  13.     code: int = 0
  14.     message: str = "success"
  15.     data: Optional[dict] = None
  16.     guard_triggered: bool = False
  17.     guard_reason: Optional[str] = None

DTO 层使用 Pydantic 进行校验，可以在进入业务逻辑之前拦截非法输入。例如，message 字段限定了最小长度和最大长度，超过范围会直接返回 422 错误。这属于第一层“被动防护”。

七、模型客户端：抽象与模拟实现

模型客户端负责与底层模型进行交互。在工程实践中，建议先定义抽象基类，再提供模拟实现和真实实现。这样可以在没有真实密钥的情况下完成全部链路调试。

  1. # app/client/base.py
  2. from abc import ABC, abstractmethod
  3. from typing import Any
  6. class BaseModelClient(ABC):
  7.     @abstractmethod
  8.     def chat(self, messages: list[dict[str, str]], **kwargs) -> str:
  9.         """发送对话消息并返回模型回复"""
  10.         pass
  13. class MockModelClient(BaseModelClient):
  14.     """模拟模型客户端，用于本地开发与测试"""
  16.     def chat(self, messages: list[dict[str, str]], **kwargs) -> str:
  17.         # 简单回显最后一条用户消息，便于链路调试
  18.         last_user_msg = ""
  19.         for msg in reversed(messages):
  20.             if msg.get("role") == "user":
  21.                 last_user_msg = msg.get("content", "")
  22.                 break
  23.         return f"[mock-reply] {last_user_msg}"

模拟客户端的行为保持极简，目的是让链路跑通，而不是模拟真实模型的所有特性。后续接入真实模型时，只需新增一个继承 BaseModelClient 的客户端类，并在配置中切换即可。

八、输入护栏：安全边界的第一道闸门

输入护栏（Input Guard）负责在用户消息进入主链路之前进行安全检查。这里的“安全检查”包括但不限于：长度校验、敏感词过滤、Prompt 注入特征识别、异常字符检测等。

  1. # app/guard/input_guard.py
  2. from dataclasses import dataclass
  5. @dataclass
  6. class GuardResult:
  7.     passed: bool
  8.     reason: str = ""
  11. class InputGuard:
  12.     BLOCKED_PATTERNS = [
  13.         "忽略以上指令",
  14.         "ignore previous instructions",
  15.         "system prompt",
  16.         "系统提示词",
  17.     ]
  19.     def check(self, message: str) -> GuardResult:
  20.         if not message or not message.strip():
  21.             return GuardResult(passed=False, reason="empty message")
  23.         if len(message) > 2000:
  24.             return GuardResult(passed=False, reason="message too long")
  26.         lowered = message.lower()
  27.         for pattern in self.BLOCKED_PATTERNS:
  28.             if pattern.lower() in lowered:
  29.                 return GuardResult(passed=False, reason=f"blocked pattern: {pattern}")
  31.         return GuardResult(passed=True)

输入护栏的逻辑应当保持独立，便于后续扩展。例如可以增加正则规则、调用外部审核接口、或者接入向量检索识别恶意输入。关键原则是：所有用户输入在进入模型之前，都必须经过这道闸门。

九、RAG 模块：最小化的检索增强链路

RAG（Retrieval-Augmented Generation，检索增强生成）的核心思路是：先从外部知识库中检索相关文本片段，再把检索结果作为上下文交给模型生成回复。本章实现一个最小化的 RAG 存储和写入接口。

  1. # app/rag/store.py
  2. import json
  3. import os
  4. from typing import List, Dict, Any
  7. class SimpleRagStore:
  8.     def __init__(self, store_path: str):
  9.         self.store_path = store_path
  10.         self.chunks: List[Dict[str, Any]] = []
  11.         self._load()
  13.     def _load(self):
  14.         if os.path.exists(self.store_path):
  15.             with open(self.store_path, "r", encoding="utf-8") as f:
  16.                 self.chunks = json.load(f)
  18.     def _save(self):
  19.         os.makedirs(os.path.dirname(self.store_path), exist_ok=True)
  20.         with open(self.store_path, "w", encoding="utf-8") as f:
  21.             json.dump(self.chunks, f, ensure_ascii=False, indent=2)
  23.     def add(self, chunk: Dict[str, Any]):
  24.         self.chunks.append(chunk)
  25.         self._save()
  27.     def search(self, query: str, top_k: int = 3) -> List[Dict[str, Any]]:
  28.         # 简化实现：按文本重叠度进行关键词匹配
  29.         query_terms = set(query.lower().split())
  30.         scored = []
  31.         for chunk in self.chunks:
  32.             text = chunk.get("text", "").lower()
  33.             overlap = len(query_terms & set(text.split()))
  34.             if overlap > 0:
  35.                 scored.append((overlap, chunk))
  36.         scored.sort(key=lambda x: x[0], reverse=True)
  37.         return [item[1] for item in scored[:top_k]]
  39.     def list_chunks(self) -> List[Dict[str, Any]]:
  40.         return self.chunks

文本切块逻辑放在 rag/ingest.py 中：

  1. # app/rag/ingest.py
  2. from typing import List
  5. def split_text(text: str, chunk_size: int = 500, overlap: int = 50) -> List[str]:
  6.     """简单固定长度切块，带重叠"""
  7.     if not text:
  8.         return []
  9.     chunks = []
  10.     start = 0
  11.     text_len = len(text)
  12.     while start < text_len:
  13.         end = min(start + chunk_size, text_len)
  14.         chunks.append(text[start:end])
  15.         if end == text_len:
  16.             break
  17.         start = end - overlap
  18.     return chunks

这里的实现刻意保持简单，目的是让 RAG 链路在本地跑通。生产环境中可以替换为向量检索方案，但这不影响整体服务架构的设计。

十、服务编排：把每一步串成完整链路

服务编排层是整个 Agent 的核心。它负责把输入护栏、RAG 检索、工具调用和模型生成按正确顺序组装起来。

  1. # app/service/chat_service.py
  2. from app.guard.input_guard import InputGuard, GuardResult
  3. from app.client.base import BaseModelClient
  4. from app.rag.store import SimpleRagStore
  5. from app.dto.chat import ChatRequest, ChatResponse
  8. class ChatService:
  9.     def __init__(self, model_client: BaseModelClient, rag_store: SimpleRagStore):
  10.         self.model_client = model_client
  11.         self.rag_store = rag_store
  12.         self.input_guard = InputGuard()
  14.     def process(self, request: ChatRequest) -> ChatResponse:
  15.         # 第一步：输入护栏
  16.         guard_result: GuardResult = self.input_guard.check(request.message)
  17.         if not guard_result.passed:
  18.             return ChatResponse(
  19.                 code=400,
  20.                 message="guard rejected",
  21.                 guard_triggered=True,
  22.                 guard_reason=guard_result.reason,
  23.             )
  25.         # 第二步：RAG 检索
  26.         rag_chunks = self.rag_store.search(request.message, top_k=3)
  27.         context_text = "\n".join([c.get("text", "") for c in rag_chunks])
  29.         # 第三步：组装消息
  30.         messages = []
  31.         if context_text:
  32.             messages.append({
  33.                 "role": "system",
  34.                 "content": f"以下是从知识库检索到的相关信息，请在回答时参考：\n{context_text}"
  35.             })
  36.         if request.user_profile:
  37.             messages.append({
  38.                 "role": "system",
  39.                 "content": f"当前用户档案：{request.user_profile}"
  40.             })
  41.         messages.append({"role": "user", "content": request.message})
  43.         # 第四步：主模型生成
  44.         reply = self.model_client.chat(messages)
  46.         return ChatResponse(
  47.             code=0,
  48.             message="success",
  49.             data={"reply": reply, "rag_chunks_used": len(rag_chunks)},
  50.         )

可以看到，服务编排层只负责“编排”，不直接实现具体业务逻辑。输入校验在 Guard 层，检索逻辑在 RAG 层，模型交互在 Client 层。这种分层让代码更易测试、更容易替换组件。

十一、HTTP 接口层：FastAPI 应用入口

接口层负责接收 HTTP 请求，把它们转换为 DTO，调用服务层，再返回 JSON 响应。

  1. # app/main.py
  2. from fastapi import FastAPI, Depends
  3. from app.config import load_config, AppConfig
  4. from app.client.base import MockModelClient
  5. from app.rag.store import SimpleRagStore
  6. from app.service.chat_service import ChatService
  7. from app.dto.chat import ChatRequest, ChatResponse
  10. app = FastAPI(title="Agent Service", version="0.1.0")
  12. _config: AppConfig = load_config()
  13. _model_client = MockModelClient()
  14. _rag_store = SimpleRagStore(_config.rag_store_path)
  15. _chat_service = ChatService(_model_client, _rag_store)
  18. @app.get("/health")
  19. def health():
  20.     return {
  21.         "status": "ok",
  22.         "model_mode": _config.model.mode.value,
  23.         "model_name": _config.model.model_name,
  24.     }
  27. @app.post("/agent/chat", response_model=ChatResponse)
  28. def agent_chat(request: ChatRequest):
  29.     return _chat_service.process(request)
  32. @app.post("/rag/ingest")
  33. def rag_ingest(text: str, chunk_size: int = 500):
  34.     from app.rag.ingest import split_text
  35.     chunks = split_text(text, chunk_size)
  36.     for i, chunk in enumerate(chunks):
  37.         _rag_store.add({"id": f"chunk_{i}", "text": chunk})
  38.     return {"code": 0, "message": "success", "data": {"chunk_count": len(chunks)}}

/health 接口返回服务状态和当前模型配置，方便运维侧快速确认服务是否正常。/agent/chat 接口承担主要对话功能。/rag/ingest 接口负责将外部文本写入 RAG 存储。

十二、测试策略：轻量验证与渐进式开发

每完成 2 到 4 个文件，就运行一次轻量验证，这一点怎么强调都不过分。推荐的验证节奏如下：

第一步：验证配置模块

在完成 config.py 后，在终端中运行：

  1. python -c "from app.config import load_config; print(load_config())"

预期输出应显示配置对象，模型模式为 mock。

第二步：验证 DTO 校验

完成 DTO 后，可以快速验证非法输入是否被拦截：

  1. python -c "from app.dto.chat import ChatRequest; ChatRequest(message='')"

应当抛出校验错误。

第三步：验证 RAG 存储

完成 RAG 模块后，测试文本切块和存储：

  1. python -c "
  2. from app.rag.ingest import split_text
  3. from app.rag.store import SimpleRagStore
  4. store = SimpleRagStore('./data/test_rag.json')
  5. for i, c in enumerate(split_text('这是一段用于测试的文本内容。' * 100)):
  6.     store.add({'id': str(i), 'text': c})
  7. print(len(store.list_chunks()))
  8. "

第四步：验证完整链路

完成服务和接口层后，启动服务并测试：

  1. uvicorn app.main:app --host 0.0.0.0 --port 8000

然后在另一个终端中测试：

  1. curl -X POST http://localhost:8000/agent/chat \
  2.   -H "Content-Type: application/json" \
  3.   -d '{"message": "你好"}'

预期返回模拟模型的回复。

十三、今日计划接口：多数据源编排示例

/agent/today 接口展示了 Agent 服务在“非纯对话”场景下的应用。它需要结合用户档案和打卡记录生成当日计划。核心流程如下：

  1. 读取用户档案 -> 读取当日打卡记录 -> 组装 Prompt -> 调用模型生成计划 -> 返回结构化结果
  1. # app/service/today_service.py
  2. from app.client.base import BaseModelClient
  3. from app.dto.today import TodayPlanRequest, TodayPlanResponse
  6. class TodayService:
  7.     def __init__(self, model_client: BaseModelClient):
  8.         self.model_client = model_client
  10.     def generate(self, request: TodayPlanRequest) -> TodayPlanResponse:
  11.         # 组装上下文
  12.         messages = [
  13.             {
  14.                 "role": "system",
  15.                 "content": "你是一个个人日程规划助手。根据用户档案和打卡记录，生成今日计划。"
  16.             },
  17.             {
  18.                 "role": "user",
  19.                 "content": (
  20.                     f"用户档案：{request.user_profile}\n"
  21.                     f"今日打卡记录：{request.check_in_records}\n"
  22.                     "请生成今日计划。"
  23.                 )
  24.             }
  25.         ]
  26.         plan_text = self.model_client.chat(messages)
  27.         return TodayPlanResponse(
  28.             code=0,
  29.             message="success",
  30.             data={"plan": plan_text}
  31.         )

这个接口的价值在于展示 Agent 服务如何编排多个输入源。后续可以将打卡记录替换为来自数据库或第三方系统 API 的数据，模型调用部分无需改动。

十四、从最小骨架到生产级服务的演进路径

本章搭建的是一个最小可运行的 Agent 服务骨架。它成功把模型调用放进了工程结构中，但要走向生产环境，还需要在以下方向持续演进：

  1. 配置管理增强：引入配置中心或加密存储，管理多环境配置。
  2. 日志与监控：增加结构化日志、请求追踪 ID、关键指标埋点。
  3. 真实模型接入：实现真实模型客户端，支持流式响应与超时重试。
  4. RAG 升级：将关键词检索替换为向量检索，引入嵌入模型与向量数据库。
  5. 工具调用规范化：定义工具调用的 JSON Schema 规范，支持多轮工具执行。
  6. 测试体系完善：补充单元测试、集成测试和契约测试，接入 CI 流程。
  7. 安全防护深化：增加输出审查、速率限制、用户认证与权限控制。

这些演进方向并不是一次性完成，而是随着业务需求逐步推进。本章的意义在于提供一个正确的起点：模型调用从一开始就处于工程化框架之中，而不是游离在业务系统之外。

十五、总结

通过本章的实践，你获得的不是一个“会说话的脚本”，而是一个可以独立运行的 Python Agent Service。它具备清晰的分层结构、明确的接口边界、基础的安全防护和可扩展的 RAG 链路。更重要的是，它遵循了渐进式开发原则——每完成几个文件就验证一次，让错误在第一时间被发现和修复。

这个最小骨架为后续接入真实模型、扩展复杂工具链、对接企业后端系统提供了可靠的地基。无论你最终选择哪家模型服务商，这套工程结构都能让你在“调通模型”之后，真正迈出“服务化”的关键一步。

最热文章