agent-">从零构建一个可运行的 Python Agent 服务：最小工程化模型调用实践

很多开发者第一次接触大模型时，往往只完成一次“调用模型并打印回复”的脚本。这样的脚本可以验证接口连通性，却很难直接进入业务系统。真正可维护的 Agent 服务，不只是调用模型，而是要把模型能力放进一个具备配置管理、输入输出约束、服务编排、安全边界和可测试性的工程结构里。

本文的目标是完成一个最小但完整的 Python Agent Service。它不依赖特定云厂商，也不要求读者预先拥有真实模型密钥。默认情况下，服务可以使用模拟链路完成开发和测试；只有显式开启真实模型配置时，才会发起真实请求。完成之后，这个服务可以被 Spring Boot 等后端系统通过 HTTP 调用，也可以独立运行与扩展。

---

一、这个服务最终长什么样

在开始写代码之前，先明确服务边界。该服务不再是一个临时脚本，而是一个长期运行的 HTTP 服务，至少包含以下能力：

1. 健康检查
   用于确认服务是否存活、当前加载了哪类模型配置、是否处于模拟模式。

2. 对话入口
   接收用户消息，先经过输入护栏，再进入检索、工具调用和主模型链路。

3. 日程生成
   基于用户档案与打卡记录，生成当日计划。

4. 轻量 RAG 摄取
   将文本切块后写入最小检索存储，为后续问答提供上下文。

从外部看，这是一个典型的 Agent 服务雏形；从内部看，它需要把配置、数据传输对象、模型客户端、服务编排、安全校验和测试反馈全部组织起来。这样后续无论是替换模型、接入真实知识库，还是扩展工具调用，都不需要推翻重来。

---

二、工程结构先行：为什么不能只写一个函数

如果只写一个 chat() 函数，短期内似乎更快，但很快就会遇到以下问题：

• 模型配置散落在代码里，切换模拟与真实模式需要改多处；
• 请求参数没有统一结构，字段校验难以复用；
• 安全边界和业务逻辑耦合在一起，后续维护成本高；
• 没有健康检查，后端无法判断服务是否可用；
• 测试只能靠人工运行脚本，无法快速定位错误来源。

因此，本章从一开始就采用“先搭骨架，再填充能力”的方式。工程结构不需要复杂，但必须清晰。一个推荐的最小结构如下：

agent-service/
├── app/
│   ├── main.py
│   ├── config.py
│   ├── dto/
│   │   └── chat.py
│   ├── clients/
│   │   └── model_client.py
│   ├── services/
│   │   ├── agent_chat.py
│   │   ├── today_plan.py
│   │   └── rag_ingest.py
│   ├── guards/
│   │   └── input_guard.py
│   └── routes/
│       ├── health.py
│       ├── agent.py
│       └── rag.py
└── tests/
    └── test_smoke.py

这个结构没有引入复杂框架，只是用模块边界把职责拆开。即使文件数量增加，每个文件只做一件事，反而更容易测试和定位问题。

---

三、配置层：把“模式”显式化

配置层的核心任务是回答一个问题：当前服务运行在什么模式下？

建议至少包含以下字段：

from dataclasses import dataclass
from typing import Optional
@dataclass
class ModelConfig:
    provider: str = "mock"
    model_name: str = "mock-model"
    api_key: Optional[str] = None
    enable_real_call: bool = False
    timeout_seconds: int = 30

这里的关键在于 enable_real_call。默认情况下，它为 False，模型客户端返回模拟结果；只有显式设置为 True 且提供有效密钥时，才会发起真实模型请求。这样做有两个好处：

1. 开发和测试环境可以无密钥运行；
2. 避免误触发计费或泄露密钥。

配置读取可以来自环境变量，也可以来自本地配置文件。为了保持简单，推荐环境变量优先，并给出一组清晰默认值。例如：

AGENT_PROVIDER=mock
AGENT_MODEL_NAME=mock-model
AGENT_API_KEY=
AGENT_ENABLE_REAL_CALL=false
AGENT_TIMEOUT_SECONDS=30

这样运维人员或后端部署时，不需要改代码，只需要调整环境变量。

---

四、DTO 与请求约束：把输入输出固定下来

HTTP 接口最容易出现的问题，就是请求字段随意、响应结构不稳定。DTO 的作用是在进入业务逻辑之前，先完成字段定义和基础校验。

例如对话请求可以定义为：

from pydantic import BaseModel, Field
class ChatRequest(BaseModel):
    message: str = Field(..., min_length=1, max_length=4000)
    user_id: str = Field(..., min_length=1, max_length=64)
    session_id: str = Field(default="default", max_length=64)
    use_rag: bool = False

响应结构同样需要稳定：

class ChatResponse(BaseModel):
    reply: str
    meta: dict = {}
    elapsed_ms: int = 0

使用 DTO 后，调用方知道应该传什么、会收到什么；服务端也可以在进入业务链路前就完成格式校验，减少无效调用进入模型层。

---

五、模型客户端：统一模拟与真实调用

模型客户端负责屏蔽底层差异。对于服务编排层来说，它只需要调用一个统一方法，比如 generate(messages) -> str，而不需要关心当前是模拟模式还是真实模式。

一个最小实现可以这样组织：

class ModelClient:
    def __init__(self, config: ModelConfig):
        self.config = config
    def generate(self, messages: list[dict]) -> str:
        if not self.config.enable_real_call:
            return self._mock_generate(messages)
        return self._real_generate(messages)
    def _mock_generate(self, messages: list[dict]) -> str:
        last = messages[-1]["content"] if messages else ""
        return f"[mock] 已收到消息：{last[:80]}"
    def _real_generate(self, messages: list[dict]) -> str:
        # 这里只描述通用流程：
        # 1. 读取 endpoint、api_key、model_name
        # 2. 构造请求体
        # 3. 设置超时与异常处理
        # 4. 解析返回文本
        raise NotImplementedError("真实模型调用需显式配置")

这样做的好处是，业务层不会因为模拟模式和真实模式切换而改变代码。测试时使用模拟客户端，生产环境切换真实客户端即可。

---

六、安全边界：输入护栏必须前置

Agent 服务接收的是用户输入，因此不能直接把原始消息交给模型或工具。输入护栏至少应包含：

• 空消息拦截；
• 超长消息截断或拒绝；
• 敏感指令识别；
• 高风险话题提示；
• 用户身份基础校验。

输入护栏适合作为独立模块存在，方便测试和调整策略。例如：

class InputGuard:
    def check(self, message: str) -> tuple[bool, str]:
        if not message or not message.strip():
            return False, "消息不能为空"
        if len(message) > 4000:
            return False, "消息过长"
        # 可扩展敏感词、注入指令检测等
        return True, "ok"

在服务编排中，应该先执行护栏检查，只有通过后才进入 RAG、工具和主模型链路。这样即使后续增加了更多工具调用，安全边界也不会被绕过。

---

七、服务编排：把多个步骤串成一个稳定流程

/agent/chat 的核心不是直接调用模型，而是编排多个步骤。最小流程可以描述为：

1. 解析请求；
2. 执行输入护栏；
3. 如果启用 RAG，进行检索并形成上下文；
4. 如果存在工具调用条件，则准备工具结果；
5. 组装消息；
6. 调用模型客户端；
7. 返回统一响应。

用伪代码表示：

def handle_chat(request: ChatRequest) -> ChatResponse:
    ok, err = guard.check(request.message)
    if not ok:
        return ChatResponse(reply=err, meta={"blocked": True})
    context = []
    if request.use_rag:
        context = rag_store.search(request.message)
    tools_result = tool_runner.run_if_needed(request.message)
    messages = build_messages(
        user_message=request.message,
        context=context,
        tool_result=tools_result,
    )
    reply = model_client.generate(messages)
    return ChatResponse(reply=reply, meta={"source": "agent"})

这种编排方式让每一步都可替换、可测试。例如 RAG 模块尚未实现时，可以先返回空上下文；工具链路未接入时，可以直接跳过，不会影响主流程。

---

八、健康检查：让服务可被运维

独立服务需要让调用方知道它是否可用。一个基础健康检查接口可以返回：

{
  "status": "ok",
  "model": "mock-model",
  "mode": "mock",
  "version": "0.1.0"
}

当服务启动失败、模型配置错误或依赖组件不可用时，健康检查应返回非 200 状态码，方便负载均衡或监控系统及时发现。

健康检查还可以承担“配置可见性”的作用。开发人员不需要登录服务器查看环境变量，直接请求接口即可知道当前服务加载了什么模型、是否处于模拟模式。

---

九、RAG 摄取：最小可用的知识注入链路

RAG 摄取接口的目标不是实现完整搜索引擎，而是建立一条“文本进入系统、被切块、可被检索”的最小链路。

一个简单实现可以包括：

class SimpleRagStore:
    def __init__(self):
        self.chunks = []
    def ingest(self, text: str, chunk_size: int = 300):
        chunks = [text[i:i+chunk_size] for i in range(0, len(text), chunk_size)]
        self.chunks.extend(chunks)
        return len(chunks)
    def search(self, query: str, top_k: int = 3):
        # 最小实现可采用关键词命中或简单相似度
        scored = []
        for chunk in self.chunks:
            score = sum(1 for w in query.split() if w in chunk)
            scored.append((score, chunk))
        scored.sort(key=lambda x: x[0], reverse=True)
        return [c for _, c in scored[:top_k]]

这个实现没有引入外部向量数据库，也没有依赖特定云服务，但它已经能够验证“摄取 → 检索 → 注入上下文”的完整链路。后续如果需要升级为向量检索，只需替换 SimpleRagStore 的实现，服务编排层几乎不需要变化。

---

十、日程生成：把业务能力包装成独立接口

/agent/today 是另一个典型 Agent 能力。它不再只是根据用户单条消息生成回复，而是结合用户档案和打卡记录生成结构化结果。

请求可以包含：

{
  "user_id": "u-1001",
  "profile": "最近在学习 Python，偏好早间深度工作",
  "checkins": [
    "08:10 完成英语阅读",
    "10:30 完成接口联调"
  ]
}

服务端将这些信息组装成提示词，再交给模型客户端。模拟模式下可以返回固定模板，真实模式下则根据模型结果调整格式。

这个接口的意义在于演示：Agent 不只是聊天机器人，它还可以基于上下文完成特定业务任务。后续可以扩展为日报生成、任务拆解、异常提醒等。

---

十一、测试策略：每完成几个文件就验证一次

本章强调“不要等整章结束才测试”。因为 Agent 服务涉及多个模块，如果最后一次性运行，错误会叠加在一起，很难判断是配置问题、结构问题还是逻辑问题。

建议按以下节奏验证：

1. 完成配置和 DTO 后，运行导入测试；
2. 完成模型客户端后，用模拟模式验证 generate；
3. 完成护栏后，验证空消息和超长消息；
4. 完成路由后，用 HTTP 客户端访问健康检查；
5. 完成 RAG 摄取后，验证文本切块和检索结果；
6. 完成对话编排后，跑一次全链路冒烟测试。

测试不一定要复杂，关键是每次改动后都有快速反馈。可以编写一个轻量测试脚本，也可以直接使用 HTTP 请求验证。对于关键模块，建议保留自动化测试文件，方便后续回归。

---

十二、从最小服务到可扩展 Agent 平台

完成上述内容后，得到的并不是一个“玩具程序”，而是一个可以继续演进的工程基座。后续可以在此基础上逐步引入：

• 更强的内容审核与安全护栏；
• 多模型切换与降级策略；
• 向量检索与知识库管理；
• 工具调用与外部 API 编排；
• 会话状态持久化；
• 日志、指标与链路追踪；
• 异步任务与并发控制。

关键点在于：每一步扩展都可以在现有模块边界内完成，而不需要重写整个服务。这也是本文强调“工程结构先行”的核心原因。

---

结语

把模型调用工程化，并不意味着一定要引入复杂框架或重型平台。相反，一个最小可运行的 Agent 服务，完全可以通过清晰的分层、显式的模式开关、稳定的 DTO、前置的安全护栏和持续的轻量验证来实现。完成这个基础版本后，开发者获得的不只是一个能回答问题的接口，而是一个可持续演进的 Agent 工程骨架。