从零构建最小可用模型调用服务：工程化落地的完整路径

一、为什么需要“最小模型调用服务”

在多数初学者示例中，模型调用往往被简化为一段“输入文本—返回结果”的脚本。然而，真实业务场景中，模型调用需要被纳入完整的工程体系：服务需要暴露明确的接口、处理异常、记录日志、实施安全策略，并支持后续迭代与集成。因此，本章的目标是构建一个最小但完整的“模型服务”（Model Service），它不是一个临时脚本，而是一个可以被其他后端模块（如 Web 应用）稳定调用的独立进程。

该服务需要具备以下核心能力：

• 健康检查：对外暴露状态接口，便于负载均衡器或监控系统确认服务存活。
• 对话接口：接收用户消息，执行输入校验、可选的知识增强流程，并返回模型响应。
• 计划生成接口：基于用户画像与历史数据（如打卡记录）生成个人计划。
• 内容注入接口：允许外部系统向内存中的知识库添加文本片段，供检索使用。
• 配置驱动：支持通过环境变量或配置文件切换模型来源（真实模型或模拟模式）。

上述设计确保了服务具备可观测性、可操作性与可测试性，为后续接入更复杂的业务逻辑奠定基础。

二、项目结构与依赖规划

在开始编码前，需要建立清晰的项目结构。本章采用分模块设计，以保持代码的职责单一。

agent-service/
├── app/
│   ├── __init__.py
│   ├── main.py            # 服务入口，负责HTTP路由与应用启动
│   ├── config.py          # 配置管理（模型源、端口、安全开关）
│   ├── dto/               # 数据传递对象（请求/响应结构）
│   │   ├── __init__.py
│   │   ├── chat.py        # 对话请求/响应模型
│   │   └── common.py      # 通用状态码与消息结构
│   ├── clients/           # 外部能力客户端（如模型API、知识库客户端）
│   │   ├── __init__.py
│   │   └── model_client.py# 模型调用客户端（支持模拟与真实两种模式）
│   ├── services/          # 业务逻辑层（编排对话流、计划生成）
│   │   ├── __init__.py
│   │   ├── chat_service.py
│   │   └── plan_service.py
│   ├── rag/               # 最小知识检索模块
│   │   ├── __init__.py
│   │   ├── store.py       # 基于内存的向量存储与检索
│   │   └── ingester.py    # 文本切块与入库逻辑
│   └── guardrails/        # 输入护栏（安全与格式校验）
│       ├── __init__.py
│       └── input_guard.py # 内容安全过滤与长度限制
├── tests/                 # 单元与集成测试
│   ├── test_health.py
│   ├── test_chat.py
│   └── test_rag.py
├── requirements.txt
└── .env.example           # 环境变量样例

依赖方面，本章仅使用两个核心第三方库：

• FastAPI：提供异步Web框架与数据校验，简化路由与DTO声明。
• pydantic：集成于FastAPI，用于定义请求/响应模型。

模型调用部分，为规避依赖真实密钥导致的测试不稳定性，我们将设计一个抽象客户端，支持“模拟模式”与“真实模式”切换。模拟模式下，返回固定的可配置响应；真实模式下，才调用外部模型API（需显式提供密钥）。

三、配置文件与环境变量管理

所有敏感信息与运行参数不应硬编码在源码中。创建 app/config.py 文件，负责读取环境变量并提供默认值。

# app/config.py
import os
from dataclasses import dataclass
@dataclass
class Settings:
    app_name: str = "Minimal Agent Service"
    host: str = os.getenv("SERVICE_HOST", "0.0.0.0")
    port: int = int(os.getenv("SERVICE_PORT", "8000"))
    # 模型相关配置
    model_provider: str = os.getenv("MODEL_PROVIDER", "mock")  # 可选: mock, real
    model_name: str = os.getenv("MODEL_NAME", "demo-model")
    api_key: str = os.getenv("REAL_MODEL_API_KEY", "")
    # 安全与限流
    max_input_length: int = int(os.getenv("MAX_INPUT_LENGTH", "2000"))
    require_input_guard: bool = os.getenv("REQUIRE_INPUT_GUARD", "true").lower() == "true"
settings = Settings()

设计说明：使用 dataclass 简化配置访问。model_provider 字段控制客户端行为，保证在无密钥环境下，服务依然可以完整运行。require_input_guard 允许在调试时暂时关闭护栏，但在生产环境应开启。

四、定义数据传输对象（DTO）

清晰的接口契约是服务可用的前提。创建 app/dto/ 相关文件。

# app/dto/common.py
from pydantic import BaseModel
from typing import Any, Optional
class ApiResponse(BaseModel):
    code: int = 0
    message: str = "success"
    data: Optional[Any] = None
class ServiceStatus(BaseModel):
    status: str
    model_config: dict
    version: str = "0.1.0"

# app/dto/chat.py
from pydantic import BaseModel
from typing import Optional
class ChatRequest(BaseModel):
    user_id: Optional[str] = "anonymous"
    message: str
    use_rag: bool = False  # 是否启用知识增强
class ChatResponse(BaseModel):
    reply: str
    source: str = "mock"  # 标识响应来源
    processed_by: list[str] = []  # 记录处理链路

DTO 的作用不仅是数据结构定义，更是接口文档的一部分。FastAPI 将根据这些模型自动生成 OpenAPI 文档，便于前端联调与测试。

五、实现模型客户端抽象层

为了支持不同运行场景，模型客户端需要抽象统一接口。创建 app/clients/model_client.py：

# app/clients/model_client.py
from abc import ABC, abstractmethod
class ModelClient(ABC):
    @abstractmethod
    async def generate(self, prompt: str, **kwargs) -> str:
        """根据提示词生成文本"""
        pass
class MockModelClient(ModelClient):
    """模拟客户端：无需外部依赖，返回预设或回声结果"""
    async def generate(self, prompt: str, **kwargs) -> str:
        # 在实际模拟中，可以返回基于prompt的确定性结果
        return f"[模拟响应] 已收到您的消息，长度为 {len(prompt)} 字符。"
class RealModelClient(ModelClient):
    """真实模型客户端：需要API密钥"""
    def __init__(self, api_key: str, model_name: str):
        if not api_key:
            raise ValueError("真实模型调用必须提供API密钥")
        self.api_key = api_key
        self.model_name = model_name
    async def generate(self, prompt: str, **kwargs) -> str:
        # 此处应接入具体模型供应商的SDK或HTTP调用
        # 为保证中立性，不特指任何供应商，仅示意通用流程
        import httpx
        async with httpx.AsyncClient() as client:
            payload = {
                "model": self.model_name,
                "prompt": prompt,
                "max_tokens": kwargs.get("max_tokens", 512),
            }
            # 注意：此处URL应由配置注入，而非硬编码
            response = await client.post(
                kwargs.get("endpoint"), 
                json=payload, 
                headers={"Authorization": f"Bearer {self.api_key}"}
            )
            response.raise_for_status()
            return response.json()["choices"][0]["text"]

工厂模式：为了根据配置创建正确的客户端，可以在客户端包内增加工厂函数：

def create_model_client(provider: str, api_key: str = "", model_name: str = "") -> ModelClient:
    if provider == "real":
        return RealModelClient(api_key=api_key, model_name=model_name)
    return MockModelClient()

此抽象层屏蔽了供应商差异，使得业务代码无需关注模型来源，便于单元测试时注入Mock对象。

六、构造输入护栏与内容安全

输入护栏是防止恶意或异常请求进入模型链路的第一道防线。创建 app/guardrails/input_guard.py：

# app/guardrails/input_guard.py
import re
from fastapi import HTTPException, status
class InputGuard:
    MAX_LENGTH = 2000
    BLOCKED_PATTERNS = [
        r"<script.*?>.*?</script>",  # 简单脚本标签
        r"javascript:",
    ]
    @classmethod
    def validate(cls, message: str) -> str:
        # 1. 长度校验
        if len(message) > cls.MAX_LENGTH:
            raise HTTPException(
                status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
                detail=f"输入长度超出限制：{cls.MAX_LENGTH} 字符"
            )
        # 2. 空白与特殊字符清理
        cleaned = message.strip()
        if not cleaned:
            raise HTTPException(status_code=400, detail="消息内容不能为空")
        # 3. 模式匹配（简单示例，生产应使用更全面的敏感词或内容审核服务）
        for pattern in cls.BLOCKED_PATTERNS:
            if re.search(pattern, cleaned, re.IGNORECASE):
                raise HTTPException(status_code=400, detail="输入包含非法内容")
        return cleaned

护栏不仅仅过滤危险词，还应进行格式规整。在实际项目中，可以接入云端内容审核类服务（通用类目），但在本章中我们以内置规则示例，强调“护栏是必须的模块”这一工程理念。

七、最小RAG存储与检索实现

RAG（检索增强生成）是提升模型回答准确性的关键技术。实现一个极简但完整的内存存储。创建 app/rag/store.py 与 app/rag/ingester.py。

# app/rag/store.py
import hashlib
from typing import List, Tuple
class SimpleVectorStore:
    """基于关键词匹配的最简RAG存储（纯内存，非向量）"""
    def __init__(self):
        self.documents = []  # 存储 (id, text, chunks)
    def add_document(self, doc_id: str, text: str, chunk_size: int = 200) -> int:
        chunks = self._split_text(text, chunk_size)
        self.documents.append({"id": doc_id, "chunks": chunks})
        return len(chunks)
    def _split_text(self, text: str, chunk_size: int) -> List[str]:
        """简单按字节/字符切块，不做语义切分"""
        if len(text) <= chunk_size:
            return [text]
        return [text[i:i+chunk_size] for i in range(0, len(text), chunk_size)]
    def search(self, query: str, top_k: int = 3) -> List[Tuple[str, str]]:
        """基于关键词重合度进行简单打分检索"""
        results = []
        for doc in self.documents:
            for chunk in doc["chunks"]:
                # 计算简单重合度
                score = sum(1 for word in query.split() if word in chunk)
                if score > 0:
                    results.append((chunk, score))
        # 按得分降序
        results.sort(key=lambda x: x[1], reverse=True)
        return [(text, score) for text, score in results[:top_k]]
store = SimpleVectorStore()

文本切块与入库接口

# app/rag/ingester.py
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from .store import store
router = APIRouter(prefix="/rag", tags=["rag"])
class IngestRequest(BaseModel):
    doc_id: str
    content: str
    chunk_size: int = 200
@router.post("/ingest")
async def ingest_text(request: IngestRequest):
    if not request.content:
        raise HTTPException(status_code=400, detail="内容不可为空")
    chunk_count = store.add_document(request.doc_id, request.content, request.chunk_size)
    return {"doc_id": request.doc_id, "chunk_count": chunk_count}

重要提示：此处提供的检索是基于关键词的朴素实现，仅用于教学演示。生产级RAG应使用向量数据库或专用向量索引（如HNSW），并使用嵌入模型进行语义编码。本章代码示意了数据流，而非性能方案。

八、服务编排层：对话服务与计划生成

业务逻辑集中在 services 层，负责串联护栏、RAG、模型客户端。

# app/services/chat_service.py
from ..guardrails.input_guard import InputGuard
from ..clients.model_client import create_model_client, ModelClient
from ..rag.store import store
from ..config import settings
class ChatService:
    def __init__(self, model_client: ModelClient):
        self.model_client = model_client
    async def process_message(self, user_id: str, message: str, use_rag: bool = False):
        # 1. 输入护栏
        cleaned_message = InputGuard.validate(message)
        # 2. 记录处理链路
        pipeline_steps = ["guard"]
        # 3. 构建上下文提示
        context = ""
        if use_rag:
            hits = store.search(cleaned_message, top_k=2)
            if hits:
                context = "\n\n相关知识片段：\n" + "\n".join([text for text, _ in hits])
                pipeline_steps.append("rag")
        # 4. 调用模型
        prompt = f"请根据以下消息进行回应。用户身份：{user_id}\n用户消息：{cleaned_message}" + context
        reply = await self.model_client.generate(prompt, user_id=user_id)
        pipeline_steps.append("model")
        return reply, pipeline_steps

计划生成逻辑类似，但输入需要结构化数据。这里我们定义一个接收打卡记录的用户模型，生成Today计划：

# app/services/plan_service.py
from ..clients.model_client import ModelClient
class PlanService:
    def __init__(self, model_client: ModelClient):
        self.model_client = model_client
    async def generate_today_plan(self, user_profile: dict, checkins: list[str]):
        # 构造格式化提示词
        checkin_text = "\n".join([f"- {c}" for c in checkins])
        prompt = f"""用户画像：{user_profile}
最近打卡记录：
{checkin_text}
请根据以上信息，为用户生成今日简要计划（限5条，每条不超过50字）。"""
        result = await self.model_client.generate(prompt)
        return result

九、主应用与接口路由装配

现在将所有组件装配到FastAPI应用中。创建 app/main.py：

from fastapi import FastAPI
from .config import settings
from .clients.model_client import create_model_client
from .services.chat_service import ChatService
from .services.plan_service import PlanService
from .dto.common import ApiResponse, ServiceStatus
from .dto.chat import ChatRequest, ChatResponse
from .rag.ingester import router as rag_router
from .guardrails.input_guard import InputGuard
app = FastAPI(title=settings.app_name, version="0.1.0")
# 依赖初始化
model_client = create_model_client(
    provider=settings.model_provider, 
    api_key=settings.api_key, 
    model_name=settings.model_name
)
chat_service = ChatService(model_client)
plan_service = PlanService(model_client)
# 路由注册
app.include_router(rag_router)
@app.get("/health", response_model=ApiResponse)
async def health_check():
    status = ServiceStatus(
        status="running",
        model_config={
            "provider": settings.model_provider,
            "model_name": settings.model_name,
            "guard_enabled": settings.require_input_guard
        }
    )
    return ApiResponse(data=status.model_dump())
@app.post("/agent/chat", response_model=ApiResponse)
async def agent_chat(req: ChatRequest):
    # 如果全局Config要求护栏开关，则强制校验（尽管ChatService内部还会校验，这里做二次兜底）
    try:
        reply, steps = await chat_service.process_message(req.user_id, req.message, req.use_rag)
    except Exception as e:
        # 生产需要日志系统，这里简化为错误信息返回
        return ApiResponse(code=500, message=f"内部错误：{str(e)}")
    data = ChatResponse(reply=reply, source=settings.model_provider, processed_by=steps)
    return ApiResponse(data=data.model_dump())
@app.post("/agent/today")
async def today_plan(user_profile: dict = {}, checkins: list[str] = []):
    # 严格场景下应使用DTO定义，这里因篇幅简化
    plan = await plan_service.generate_today_plan(user_profile, checkins)
    return ApiResponse(data={"plan": plan})

可见性设计：/health 接口返回模型配置信息，方便运维侧快速确认服务使用的模型源。processed_by 字段记录了消息的处理路径，帮助开发者诊断是否经过RAG。

十、测试策略：分阶段验证

工程实践要求在开发过程中持续验证，而非一次性写完所有代码调试。本章建议的验证顺序如下：

1. 单模块测试：完成 config.py 后，立即打印配置确认加载。
2. 客户端测试：完成 model_client.py 后，运行一个简短的 Async 测试，确认Mock客户端返回正确。
3. RAG模块测试：写入 store 后，用Python交互式环境测试 add_document 和 search 功能。
4. 服务集成测试：启动FastAPI服务后，通过 curl 命令验证各接口。

以下是服务启动与验证命令示例：

# 启动开发服务器（使用Mock模型）
uvicorn app.main:app --reload
# 终端1：验证健康检查
curl http://127.0.0.1:8000/health
# 预期返回包含 status: "running" 和模型配置
# 终端2：验证对话接口
curl -X POST http://127.0.0.1:8000/agent/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "你好，今天天气如何？", "use_rag": false}'
# 预期返回模拟响应
# 终端3：验证RAG接口
curl -X POST http://127.0.0.1:8000/rag/ingest \
  -H "Content-Type: application/json" \
  -d '{"doc_id": "doc1", "content": "深度学习是机器学习的一个分支，基于神经网络。"}'
# 返回入库块数
# 再请求对话并开启RAG
curl -X POST http://127.0.0.1:8000/agent/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "什么是深度学习？", "use_rag": true}'

测试对齐真实模型：若开发者持有真实模型API密钥，必须通过设置环境变量 MODEL_PROVIDER=real 及 REAL_MODEL_API_KEY=xxx 后重启服务，并确保 REQUIRE_INPUT_GUARD=true 已开启。默认情况下，服务绝不对外发起真实API调用，防止误触发费用或泄露密钥。

十一、工程化要点与扩展方向

本章构建的服务虽然“最小”，但覆盖了工程化服务的核心要件：

• 可配置性：所有环境依赖均通过外部变量注入。
• 安全边界：显式实现输入校验与内容过滤，支持开关。
• 可观测性：通过 /health 与响应体中的链路追踪字段，提供基础可观测能力。
• 抽象隔离：模型客户端与业务逻辑解耦，便于替换实现或添加缓存、重试等能力。

未来的优化方向：

• 将内存RAG存储替换为持久化向量数据库。
• 引入日志模块与分布式追踪（如统一日志框架）。
• 为 agent/chat 增加超时控制与熔断器。
• 增加单元测试文件（本章未展开，但项目结构预留了tests/目录），使用 pytest 和模拟对象进行覆盖。

结语

从“单次模型调用”到“可持续演进的模型服务”，关键在于工程化思维的建立。通过本章实践，读者应已掌握构建一个RESTful模型服务的基础能力，并理解各模块的职责边界。下一章可以在此基础上，探索如何将工具调用、多轮对话状态管理或更复杂的上下文压缩机制集成进来，逐步构筑一个具备生产价值的智能体服务。