AI Agent 工程化实战：从最小工具调用循环到生产级系统

一、Agent 不是"会聊天的机器人"

很多团队把 Agent 理解成"更聪明的聊天机器人"，这个认知偏差会导致整个技术方案走偏。Agent 的本质是能感知目标、规划步骤、调用工具、根据反馈迭代的智能体。专业开发的核心，不是把 Prompt 写得花哨，而是把不确定性关进工程框架：状态可追踪、工具可管控、失败可恢复、成本可约束。

一个可落地的 Agent 通常包含四层：规划层负责把用户目标拆成子任务并决定下一步；记忆层承载短期对话上下文和长期向量检索；工具层对接函数、API、数据库、代码执行沙箱；执行循环则是"LLM 决策 → 调用工具 → 回填观察 → 继续推理"的闭环，直到产出最终答案。

主流框架有 LangChain/LangGraph、AutoGen、CrewAI、OpenAI Agents SDK 等，但无论用哪个，底层都是 ReAct 循环的变体。先吃透这个循环，再谈框架。

二、代码实战：从零实现一个工具调用 Agent

下面用 OpenAI 兼容 SDK 实现一个最小可运行的 Agent，包含计算器和天气查询两个工具：

import json, os, re
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
TOOLS = [
    {"type": "function", "function": {
            "name": "calculator",
                    "description": "计算数学表达式，例如 (12+8)*3",
                            "parameters": {"type": "object",
                                                   "properties": {"expression": {"type": "string"}},
                                                                          "required": ["expression"]}}},
                                                                              {"type": "function", "function": {
                                                                                      "name": "get_weather",
                                                                                              "description": "查询指定城市天气",
                                                                                                      "parameters": {"type": "object",
                                                                                                                             "properties": {"city": {"type": "string"}},
                                                                                                                                                    "required": ["city"]}}},
                                                                                                                                                    ]
def calculator(expression: str) -> str:
    if not re.fullmatch(r"[0-9+\-*/().\s]+", expression):
            return "表达式包含不允许的字符"
                try:
                        return str(eval(expression, {"__builtins__": {}},))
                            except Exception as e:
                                    return f"计算失败: {e}"
def get_weather(city: str) -> str:
    data = {"北京": "晴，25℃", "上海": "多云，28℃", "深圳": "阵雨，30℃"}
        return data.get(city, f"暂无{city}的天气数据")
TOOL_MAP = {"calculator": calculator, "get_weather": get_weather}

def run_agent(user_input: str, max_steps: int = 5):
    messages = [{"role": "system", "content": "你是严谨助手。需要计算或天气时调用工具，不要编造。"},
                    {"role": "user", "content": user_input}]
                        for _ in range(max_steps):
                                resp = client.chat.completions.create(
                                            model="gpt-4o-mini", messages=messages, tools=TOOLS,
                                                        tool_choice="auto", temperature=0.2)
                                                                msg = resp.choices[0].message
                                                                        messages.append(msg)
                                                                                if not msg.tool_calls:
                                                                                            return msg.content
                                                                                                    for call in msg.tool_calls:
                                                                                                                fn = call.function.name
                                                                                                                            args = json.loads(call.function.arguments)
                                                                                                                                        result = TOOL_MAP[fn](**args)
                                                                                                                                                    messages.append({"role": "tool", "tool_call_id": call.id, "content": result})
                                                                                                                                                        return "达到最大步数，任务未完成"
                                                                                                                                                        ```
这段代码展示了 Agent 的基本闭环。注意几个生产化要点：`eval` 要替换为安全计算库；模拟天气要换成真实 API；`max_steps` 必须设置，否则模型可能陷入"调工具—不满意—再调"的死循环，把 Token 烧光。

## 三、生产级关键点：六个必答题

从 Demo 到生产，不是把代码搬到服务器上，而是要回答六个问题：

**状态管理。** 用 LangGraph 这类框架把循环建模为状态图，支持中断、恢复、人工审批。会话上下文存 Redis，长任务支持断点续传。没有状态管理的 Agent 一重启就"失忆"。

**工具安全。** 工具必须最小权限。代码执行要放沙箱，禁止直接执行用户输入；所有工具调用记录审计日志。能查的、能写的、能删的要分级，危险操作前请求确认。这是 Agent 最容易出安全事故的地方。

**可观测性。** 记录每步的 prompt、token、耗时、工具入参出参，接入 LangSmith 或 OpenTelemetry。没有 trace，线上出问题只能靠猜。

**成本控制。** 设置最大步数、最大 token、单次预算；小模型做路由和简单任务，大模型做复杂推理。多模型路由是成本优化的第一杠杆。

**评估体系。** 用固定测试集评估任务完成率、工具调用准确率、幻觉率，而不是只看"回答像不像人"。评测集要覆盖正常路径和边界路径。

**合规红线。** 遵守平台条款、隐私法规、开源许可证；不开发用于欺诈、攻击、骚扰的 Agent；涉及 AI 生成内容的场景按需标注。

## 四、工程化架构：分层解耦的 Agent 工厂

生产级 Agent 适合采用分层架构，把 API 层、编排层、记忆层、工具层解耦。每层通过依赖注入连接，便于单元测试和替换实现。

```python
class AgentOrchestrator:
    def __init__(self, tools, memory, redis_client):
            self.tools = {t.name: t for t in tools}
                    self.llm = ChatOpenAI(model="gpt-4o", temperature=0.2, streaming=True)
                            self.memory = memory
                                    self.max_iterations = 5
    async def run(self, user_input, session_id, stream_callback=None):
            history = self._get_short_term(session_id)
                    memories = await self.memory.retrieve_relevant(user_input)
                            system_prompt = self._build_system_prompt(memories)
                                    messages = [SystemMessage(content=system_prompt)] + history + [HumanMessage(content=user_input)]
                                            for step in range(self.max_iterations):
                                                        response = await self.llm.ainvoke(
                                                                        messages, tools=[t.to_openai_schema() for t in self.tools.values()],
                                                                                        tool_choice="auto")
                                                                                                    messages.append(response)
                                                                                                                if not response.tool_calls:
                                                                                                                                return response.content
                                                                                                                                            tool_results = await self._execute_tools(response.tool_calls)
                                                                                                                                                        for tc, result in zip(response.tool_calls, tool_results):
                                                                                                                                                                        messages.append({"role": "tool", "tool_call_id": tc["id"], "content": result})
                                                                                                                                                                                return "达到最大步数"
                                                                                                                                                                                ```
记忆系统采用"短期 + 长期"双轨：短期用 Redis 存当前会话的滑动窗口，长期用向量库存历史问答，每次推理前按相关性注入。双轨的好处是既保持对话连贯，又不让上下文无限膨胀。

## 五、容错与降级设计

LLM 接口不稳定是常态，容错设计要分层：工具调用设置独立超时；可重试错误（429、5xx）用指数退避重试；连续失败时降级——LLM 不可用返回缓存答案，工具失败返回预设兜底响应。重试放大是个容易被忽视的坑：同一个请求不要让 SDK、网关、队列三层都重试，否则一次上游抖动会放大成几倍的请求洪峰。

## 六、评估体系：用数据说话

很多团队把 Agent 上线后只看"回答像不像人"，这是最大的误区。评估要围绕三个可量化指标：任务完成率（测试集里成功完成的比例）、工具调用准确率（该调的工具调对了多少次）、幻觉率（编造事实的比例）。评测集要覆盖三类用例：正常路径、边界输入（空输入、超长输入、模糊指令）、恶意输入（提示注入尝试）。

评估跑通之后，每次改 Prompt、换模型、调参数，都先过一遍评测集再上线。这一步投入不大，但能拦住绝大多数"改坏了没人发现"的回归问题。我见过太多项目死在"上线两周后质量悄悄下降"——没有评测基线，你根本不知道是哪次改动引起的。

## 七、总结与建议

大模型 Agent 的专业开发，本质是把 LLM 的不确定性封装进可观测、可控制、可恢复的工程系统。建议的成长路线：先手写一个最小工具调用循环，理解 ReAct 的本质；再引入框架做状态管理和多 Agent 协作；最后补齐观测、评估、成本治理。代码本身不复杂，难的是对边界、成本和安全的持续把控。与其看十篇概念文章，不如今天就写一个能跑的循环——从最小闭环开始，逐步长出生产级的血肉。