从零构建 AI Agent 全栈实战：从模型集成到生产部署

2026/9/282 分钟阅读

从零构建 AI Agent 全栈实战：从模型集成到生产部署

我搭过几个能跑通的 AI Agent，没用大框架，就靠几层小抽象。最常卡住的地方不是模型调不动，而是消息格式对不上、工具返回乱码、用户等三秒就关页面——这些事，文档里不写，但上线第一天就撞上。

模型集成与统一接口

OpenAI、Claude、本地 Ollama，每个 API 返回的字段都像方言。有带 choices[0].message.content 的，有直接扔 response.text 的，还有把 token 数藏在 headers 里的。我在业务逻辑里硬写三套调用？试过一次，改个 system prompt 就得翻三个文件。

从零构建 AI Agent 全栈实战：从模型集成到生产部署 - 插图 1

所以先切一刀：只留一个 chat() 方法。其他全塞进适配器里。

from abc import ABC, abstractmethod
from typing import List, Dict

class LLMProvider(ABC):
    @abstractmethod
    async def chat(self, messages: List[Dict[str, str]]) -> str:
        pass

    @abstractmethod
    def get_usage(self) -> Dict:
        pass

class OpenAIAdapter(LLMProvider):
    def __init__(self, api_key: str):
        self.client = OpenAI(api_key=api_key)

    async def chat(self, messages: List[Dict[str, str]]) -> str:
        response = await asyncio.to_thread(
            self.client.chat.completions.create,
            model="gpt-4", # 或 gpt-3.5-turbo
            messages=messages
        )
        return response.choices[0].message.content

加个新模型？建个新类，两分钟。想让客服任务走 Qwen，数据分析走 GPT-4？加个路由逻辑就行，不用动业务主干。

实现 ReAct 推理循环

模型自己瞎猜不如让它“边想边做”。ReAct 不是炫技，是救命——输出里带 "action": "search" 和 "action_input": "2024年Q2财报"，比纯文本好 debug 十倍。

但模型总爱自由发挥。我吃过亏：它在 tool call 里混进一句“顺便查下天气”，JSON 直接 parse 失败，整个对话卡死。

现在强制两点：系统提示词里写死 JSON Schema，解析层用 Pydantic 校验。错一个字段，就丢回去重试，不往下传。

async def run_agent_loop(agent_state: AgentState, tools: List[Tool]) -> str:
    while not agent_state.is_completed():
        # 1. 获取模型响应
        response = await llm_client.chat(agent_state.history)
        
        # 2. 解析工具调用
        tool_call = parse_tool_call(response)
        if not tool_call:
            return response # 最终回答
            
        # 3. 执行工具并更新状态
        tool_result = await execute_tool(tool_call, tools)
        agent_state.add_message("tool", tool_result)
        agent_state.add_message("assistant", response)
        
    return agent_state.final_answer

循环本身很短。难的是每一步怎么兜住——比如工具超时了怎么办？返回空字符串还是抛异常？这些细节，才是 Agent 能不能活过第一天的关键。

生产部署的关键考量

上线前我测过延迟：同步调用 + 完整上下文 + 一次工具执行，平均 2.8 秒。用户点完发送，盯着转圈超过三秒，手指就开始往右滑。

必须流式输出。不是为了酷，是怕用户以为挂了。

另一个坑是 history 越攒越多。第 15 轮对话，光 message list 就占掉 12K token。GPT-4 Turbo 帐户还行，换成本地 4B 模型？直接爆 context。

我的解法很土：滑动窗口砍老消息，再加个轻量摘要——比如把前三轮“查订单→确认地址→生成运单”压缩成一句“已完成订单 12345 的发货准备”。不完美，但比截断强。

状态存哪？Redis。扩机器时，会话不丢，也不用改代码。K8s 里起五个 pod，它们共享同一份 session cache。

下一步建议

RAG 我试过两次。第一次拿全文本硬喂，检索结果全是噪音；第二次加了 chunk size 控制和 rerank，准确率才上来。Vector DB 也是，选了 Chroma，但数据量一过百万，就得调相似度阈值——这些坑，下次细聊。

从零构建 AI Agent 全栈实战：从模型集成到生产部署