agent-">基于最小模型调用构建 Python Agent 工程化服务实践指南

一、引言：为什么需要”工程化”而非”一次性调用”

在智能应用开发的早期阶段，开发者往往从一段简单的脚本开始——初始化一个模型客户端、构造一条提示词、获取返回结果。这种做法在原型验证阶段完全合理，但一旦进入生产环境或需要被后端系统（如 Java Spring Boot 微服务）稳定调用时，单次调用模式就会暴露出诸多问题：缺乏统一的配置管理、没有输入验证与安全护栏、无法支撑多轮对话与工具链路、缺少健康探测接口、也没有标准化的数据交换格式。

本章的核心目标，是将”让模型说一句话”升级为”让模型在一个完整工程结构中可靠地工作”。最终产出是一个具备以下能力的 Python Agent Service：

• 健康检查接口（/health）：返回服务运行状态与当前模型配置信息，便于运维监控。
• 对话接口（/agent/chat）：接收用户消息，经过输入护栏后依次进入 RAG 检索、工具调用和主模型推理链路。
• 日程生成接口（/agent/today）：基于用户档案与历史打卡数据，生成当日计划。
• RAG 数据摄入接口（/rag/ingest）：接收文本并完成切块后存入最小 RAG 存储。

值得注意的是，默认测试阶段无需真实的模型 API 密钥，但若需接入真实大模型进行端到端验证，则必须显式开启并配置对应凭证。

二、工程结构设计：从目录到职责划分

一个合格的 Agent Service 应当具备清晰的分层架构。推荐的目录结构如下：

agent_service/
├── config/
│   └── settings.py          # 统一配置管理
├── dto/
│   ├── request.py           # 请求数据传输对象
│   └── response.py          # 响应数据传输对象
├── client/
│   └── model_client.py      # 模型客户端封装
├── service/
│   ├── agent_service.py     # 核心编排逻辑
│   ├── rag_service.py       # RAG 链路服务
│   └── today_service.py     # 日程生成服务
├── store/
│   └── rag_store.py         # 最小 RAG 存储实现
├── api/
│   └── routes.py            # HTTP 路由与控制器
├── main.py                  # 服务入口
└── tests/
    └── test_basic.py        # 轻量验证测试

每一层职责明确：config 负责环境变量与参数加载；dto 定义数据边界；client 隔离模型 SDK 依赖；service 承载业务编排；store 管理向量或文本索引；api 暴露 HTTP 能力。

三、配置管理：让服务可移植

配置文件应集中管理模型端点、API 密钥开关、超时参数、默认模型标识等。使用环境变量加载策略，既能在本地开发时灵活切换，也能在容器化部署时通过注入方式统一管理。

# config/settings.py
import os
class Settings:
    MODEL_ENDPOINT: str = os.getenv("MODEL_ENDPOINT", "http://localhost:8000/v1")
    MODEL_API_KEY: str = os.getenv("MODEL_API_KEY", "")
    MODEL_ENABLED: bool = os.getenv("MODEL_ENABLED", "false").lower() == "true"
    DEFAULT_MODEL: str = os.getenv("DEFAULT_MODEL", "default-model")
    REQUEST_TIMEOUT: int = int(os.getenv("REQUEST_TIMEOUT", "30"))
    RAG_CHUNK_SIZE: int = int(os.getenv("RAG_CHUNK_SIZE", "512"))
    RAG_CHUNK_OVERLAP: int = int(os.getenv("RAG_CHUNK_OVERLAP", "50"))

这种设计的关键在于：MODEL_ENABLED 默认为 false，确保默认测试不会意外调用真实模型；只有显式设置为 true 时才进入真实推理链路。

四、DTO 定义：数据交换的契约

数据传输对象（Data Transfer Object）是服务间通信的契约。每一个接口都应有对应的请求和响应结构，字段类型明确、可序列化。

# dto/request.py
from pydantic import BaseModel, Field
from typing import Optional
class ChatRequest(BaseModel):
    user_id: str = Field(..., description="用户唯一标识")
    message: str = Field(..., min_length=1, max_length=4096, description="用户输入消息")
    session_id: Optional[str] = Field(None, description="会话标识，支持多轮")
class TodayRequest(BaseModel):
    user_id: str = Field(..., description="用户唯一标识")
    date: Optional[str] = Field(None, description="目标日期，默认当天")
class IngestRequest(BaseModel):
    text: str = Field(..., min_length=1, description="待摄入的原始文本")
    source: Optional[str] = Field(None, description="数据来源标识")

响应对象同理，需包含状态码、结果数据、错误信息等标准字段，便于调用方统一解析。

五、模型客户端封装：隔离与可替换

模型客户端层的职责是将具体模型 API 调用细节封装起来，对外暴露统一的 chat、embed 等方法。这样当底层模型提供商发生变更时，只需替换客户端实现，上层服务逻辑不受影响。

# client/model_client.py
import httpx
from config.settings import Settings
class ModelClient:
    def __init__(self):
        self.endpoint = Settings.MODEL_ENDPOINT
        self.api_key = Settings.MODEL_API_KEY
        self.enabled = Settings.MODEL_ENABLED
    def chat(self, messages: list[dict]) -> str:
        if not self.enabled:
            return "[模拟响应] 模型未启用，请设置 MODEL_ENABLED=true"
        headers = {"Authorization": f"Bearer {self.api_key}"}
        payload = {
            "model": Settings.DEFAULT_MODEL,
            "messages": messages,
            "max_tokens": 1024,
        }
        with httpx.Client(timeout=Settings.REQUEST_TIMEOUT) as client:
            resp = client.post(f"{self.endpoint}/chat/completions", json=payload, headers=headers)
            resp.raise_for_status()
            return resp.json()["choices"][0]["message"]["content"]

六、输入护栏：安全边界的第一道防线

在 Agent 服务中，用户输入直接参与模型推理，因此必须设置护栏机制：

1. 长度限制：防止超长输入导致 token 超限或拒绝服务。
2. 内容过滤：拦截明显的注入攻击或恶意指令。
3. 格式校验：确保必填字段完整、类型正确。

护栏逻辑应在服务编排层的最前端执行，任何不合规请求直接返回错误响应，不进入后续链路。

七、RAG 最小存储与摄入链路

RAG（检索增强生成）是提升 Agent 回答质量的关键手段。最小 RAG 存储不需要复杂的向量数据库，可以先用内存中的分块索引实现原型。

# store/rag_store.py
from typing import List, Dict
from config.settings import Settings
class RAGStore:
    def __init__(self):
        self.chunks: List[Dict] = []
    def ingest(self, text: str, source: str = "unknown") -> int:
        chunk_size = Settings.RAG_CHUNK_SIZE
        overlap = Settings.RAG_CHUNK_OVERLAP
        chunks = []
        start = 0
        while start < len(text):
            end = min(start + chunk_size, len(text))
            chunks.append({
                "id": len(self.chunks) + len(chunks),
                "text": text[start:end],
                "source": source,
            })
            start = end - overlap if end < len(text) else end
        self.chunks.extend(chunks)
        return len(chunks)
    def search(self, query: str, top_k: int = 3) -> List[Dict]:
        # 简单关键词匹配作为原型，生产环境应替换为向量相似度检索
        scored = []
        for chunk in self.chunks:
            if query.lower() in chunk["text"].lower():
                scored.append(chunk)
        return scored[:top_k]

/rag/ingest 接口接收文本后调用 ingest 方法完成切块存储，后续 /agent/chat 中的 RAG 环节则调用 search 获取相关片段作为上下文注入提示词。

八、服务编排：将护栏、RAG、工具、主模型串成链路

核心编排逻辑是 Agent 的”大脑”。推荐采用管道式设计：输入 → 护栏 → RAG 检索 → 工具调用（如有）→ 主模型推理 → 输出。

# service/agent_service.py
from dto.request import ChatRequest
from client.model_client import ModelClient
from store.rag_store import RAGStore
class AgentService:
    def __init__(self):
        self.model_client = ModelClient()
        self.rag_store = RAGStore()
    def chat(self, req: ChatRequest) -> str:
        # 1. 输入护栏已由 Pydantic 在 DTO 层完成
        # 2. RAG 检索
        context_chunks = self.rag_store.search(req.message)
        context_text = "\n".join([c["text"] for c in context_chunks])
        # 3. 构造提示词
        system_prompt = f"你是一个智能助手。参考以下上下文回答用户问题。\n上下文：{context_text}"
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": req.message},
        ]
        # 4. 调用主模型
        return self.model_client.chat(messages)

/agent/today 接口则需要额外加载用户档案和打卡数据，构造专属提示词后调用模型生成计划，逻辑类似但上下文构造更复杂。

九、健康检查与 HTTP 路由

使用轻量 Web 框架暴露接口，健康检查接口应返回 JSON 格式的状态信息，包括服务版本、模型配置状态、存储健康度等。

# api/routes.py
from fastapi import FastAPI
from dto.request import ChatRequest, TodayRequest, IngestRequest
from service.agent_service import AgentService
from service.rag_service import RAGService
from service.today_service import TodayService
app = FastAPI(title="Agent Service")
agent_svc = AgentService()
rag_svc = RAGService(agent_svc.rag_store)
today_svc = TodayService()
@app.get("/health")
def health():
    return {
        "status": "ok",
        "model_enabled": agent_svc.model_client.enabled,
        "model": "default-model",
        "rag_chunks_count": len(agent_svc.rag_store.chunks),
    }
@app.post("/agent/chat")
def do_chat(req: ChatRequest):
    return {"reply": agent_svc.chat(req)}
@app.post("/rag/ingest")
def do_ingest(req: IngestRequest):
    count = rag_svc.ingest(req.text, req.source)
    return {"status": "ok", "chunks": count}

十、轻量验证策略：小步快跑，避免错误堆积

本章强调一个重要工程原则：每完成 2 到 4 个文件就运行一次轻量验证，而非等整章结束才测试。具体做法：

1. 配置层完成后：运行 Python 导入测试，确认环境变量加载无报错。
2. DTO 层完成后：用无效数据构造请求对象，验证 Pydantic 校验能正确抛出异常。
3. 客户端层完成后：在 MODEL_ENABLED=false 下调用 chat，确认返回模拟响应。
4. 服务层完成后：启动 HTTP 服务，用 curl 调用 /health 确认接口可达。

这种渐进式验证能将错误定位到具体步骤，极大降低调试成本。

十一、总结与展望

本章完成的不是一个”能调用模型”的脚本，而是一个具备工程骨架的 Agent Service：配置集中管理、数据契约明确、客户端可替换、链路可编排、护栏有边界、接口可监控。这套结构是后续接入更复杂工具链（如函数调用、多 Agent 协作）、接入百度智能云等平台的大模型服务、以及与 Spring Boot 等后端系统对接的坚实基础。下一章可在此骨架上扩展真实模型接入、持久化存储替换、以及完整的单元测试与集成测试体系。