AI Agent 开发实战：从零构建可用的 Agent 工作流 — 核心在于"把大模型的思考能力装进可控的流水线"
我在为公司搭建内部客服助手时发现，直接调 GPT API 根本不可用——模型自由发挥太多，输出不稳定，上下文一长就失忆。最终落地的是一个基于 ReAct 模式 + 工具注册 + 状态管理 的 Agent 框架，这篇文章完整记录了从方案选型到生产部署的全过程。
问题是什么
年初接到一个需求：给公司客服团队做一个 AI 助手，能根据知识库回答用户问题、查询订单状态、发起退款流程。听起来不就是调一下 ChatGPT API 吗？真做起来才发现三个核心痛点：
我需要一个架构，让 LLM 的"思考"有边界、可观测、能落地。
解决思路
调研了三个主流方案：
| 方案 | 核心思路 | 复杂度 | 适合场景 |
|---|---|---|---|
| LangChain | 封装好的 Agent 框架，自带工具链 | 中高 | 快速原型，但出问题难排查 |
| Semantic Kernel | 微软的 AI 编排框架 | 中 | .NET 生态友好，Python 支持一般 |
| 自建 ReAct 循环 | 自己写 Prompt + Tool Registry + Loop | 低 | 理解原理，完全可控 |
我选了 自建 ReAct 循环。原因很简单：LangChain 的 Agent 是个黑盒，出了问题你连它在"想什么"都不知道。自建虽然要多写几百行代码，但每个环节的可观测性完全掌握在自己手里。
核心思路就三句话：
操作步骤
步骤 1：设计 Tool Registry — Agent 的"手和脚"
Agent 要能干活，必须有工具。我先定义了一套统一的工具接口：
from abc import ABC, abstractmethod
from typing import Dict, Any
from pydantic import BaseModel
class ToolSpec(BaseModel):
"""工具声明——告诉 LLM 这个工具是干什么的"""
name: str
description: str
parameters: Dict[str, Any] # JSON Schema 格式
class Tool(ABC):
"""工具的基类"""
spec: ToolSpec
@abstractmethod
def run(self, **kwargs) -> str:
"""执行工具，返回文本结果给 LLM"""
pass
每个工具都包含两部分：声明（告诉 LLM 这个工具的存在和用法）和实现（实际干活）。比如一个查订单的工具：
class QueryOrderTool(Tool):
spec = ToolSpec(
name="query_order",
description="根据订单号查询订单状态和详细信息",
parameters={
"type": "object",
"properties": {
"order_id": {"type": "string", "description": "订单号"}
},
"required": ["order_id"]
}
)
def run(self, order_id: str) -> str:
# 实际调用订单 API
data = requests.get(f"https://api.xxx.com/orders/{order_id}")
return f"订单 {order_id} 状态: {data['status']}, 金额: {data['amount']}"
注意: ToolSpec 的
parameters一定要用 JSON Schema 格式。我一开始用自由文本描述，结果 LLM 经常传错参数类型，改成严格 schema 后问题解决了 90%。
步骤 2：构建 System Prompt — Agent 的"大脑说明书"
这是整个 Agent 最核心的部分。System Prompt 不是写一段话告诉 LLM"你要友好"，而是要给它一套可执行的行动规则：
SYSTEM_PROMPT = """你是一个客服助手。请严格按照以下流程工作：
## 可用工具
{tool_descriptions}
## 工作模式
每次回复必须遵循以下格式：
问题分析：<简述你对用户问题的分析>
思考：<根据当前信息，决定下一步动作>
动作：<调用的工具名>
参数：<JSON 格式的参数>
---
如果你已经得到最终答案，或者用户的问题不需要调用工具，直接输出：
最终回答：<你的回答>
## 关键规则
1. 每次只调用一个工具，等待结果后再决定下一步
2. 如果工具返回错误，尝试换个参数重新调用，最多重试 2 次
3. 如果用户要求查询隐私信息（密码、身份证号等），拒绝并说明原因
4. 当上下文超过 5 轮对话时，主动总结关键信息"""
{tool_descriptions} 是动态生成的，遍历所有注册的工具生成统一的描述文本：
def _build_tool_descriptions(tools: Dict[str, Tool]) -> str:
lines = []
for name, tool in tools.items():
s = tool.spec
params_desc = "
".join(
f" - {p}: {info.get('description', '')}"
for p, info in s.parameters.get("properties", {}).items()
)
lines.append(f"### {s.name}
{s.description}
参数:
{params_desc}")
return "
".join(lines)
步骤 3：实现 ReAct Loop — Agent 的"引擎"
核心循环就是一个 while True：把当前消息拼进上下文，让 LLM 生成回复，解析回复内容，执行工具，把结果追加回去，重复。
class Agent:
def __init__(self, tools: List[Tool], model: str = "gpt-4"):
self.tools = {t.spec.name: t for t in tools}
self.model = model
self.messages = []
def run(self, user_input: str) -> str:
# 初始化消息
self.messages = [
{"role": "system", "content": SYSTEM_PROMPT.format(
tool_descriptions=_build_tool_descriptions(self.tools)
)},
{"role": "user", "content": user_input}
]
max_steps = 10 # 防止无限循环
for step in range(max_steps):
# 1. 调用 LLM
response = self._call_llm(self.messages)
content = response.choices[0].message.content
# 2. 检查是否最终回答
if content.startswith("最终回答："):
return content.replace("最终回答：", "").strip()
# 3. 解析工具调用
action = self._parse_action(content)
if action is None:
# LLM 输出格式不对，给个纠正提示
self.messages.append({
"role": "user",
"content": "格式错误，请严格按照 '动作：<工具名>
参数：<JSON>' 格式输出"
})
continue
# 4. 执行工具
try:
tool = self.tools[action["name"]]
result = tool.run(**action["parameters"])
except Exception as e:
result = f"工具执行出错: {str(e)}"
# 5. 把结果追加到对话中
self.messages.append({
"role": "user",
"content": f"工具 {action['name']} 返回结果：
{result}
请根据结果继续。"
})
return "抱歉，处理超时，请稍后重试。"
踩坑记录: 我一开始把工具结果用
role: "tool"塞进去，结果 GPT-4 在某些版本对这个 role 的支持不一致。最后改用role: "user"+ 明确的前缀标记，稳定多了。
步骤 4：增加状态管理 — 解决"失忆"问题
纯靠 Prompt 的 Agent 会在长对话中丢失关键信息。我加了一个显式的 Memory 模块：
class ConversationMemory:
def __init__(self, max_tokens: int = 3000):
self.max_tokens = max_tokens
self.summary = ""
self.key_info = {} # 关键信息：用户ID、订单号等
def extract_key_info(self, text: str):
"""从对话中提取关键信息"""
# 简单的规则提取，生产环境可以用 NER 模型
patterns = {
"user_id": r"用户[I云号]?[：:]\s*(\w+)",
"order_id": r"订单[号]?[：:]\s*(\w+)",
"phone": r"1[3-9]\d{9}",
}
for key, pattern in patterns.items():
match = re.search(pattern, text)
if match:
self.key_info[key] = match.group(1)
def build_context(self) -> str:
"""生成压缩后的上下文"""
parts = []
if self.summary:
parts.append(f"对话摘要（{self.summary}）")
if self.key_info:
info_str = "；".join(f"{k}={v}" for k, v in self.key_info.items())
parts.append(f"已确认信息：{info_str}")
return "
".join(parts)
然后在 Agent Loop 里，每 3 轮调用一次 LLM 做摘要压缩：
if step > 0 and step % 3 == 0:
summary_prompt = f"请用一句话总结对话经过，提取关键信息:
{self.messages}"
summary = self._call_llm_simple(summary_prompt)
memory.summary = summary
memory.extract_key_info(summary)
# 把旧消息压缩，只保留最近 2 轮和摘要
self.messages = self.messages[:1] + [ # system prompt
{"role": "user", "content": f"【对话摘要】{memory.build_context()}"}
] + self.messages[-4:] # 最近两轮对话
步骤 5：加可观测性 — 让 Agent 的思考过程可见
客服主管不信任 AI 的原因是"我不知道它怎么得出这个结论的"。我加了一个简单的 Thought Logger：
class ThoughtLogger:
def __init__(self):
self.steps = []
def log(self, step: int, action: str, reasoning: str, result: str):
self.steps.append({
"step": step,
"action": action,
"reasoning": reasoning,
"result": result[:200], # 截断过长内容
"timestamp": datetime.now().isoformat()
})
def to_html(self) -> str:
"""生成可展示的推理过程"""
html = "<div class='thought-log'>"
for s in self.steps:
html += f"""
<div class='step'>
<div class='step-header'>Step {s['step']}: {s['action']}</div>
<div class='step-reasoning'>🤔 {s['reasoning']}</div>
<div class='step-result'>✅ {s['result']}</div>
</div>
"""
html += "</div>"
return html
把这个 HTML 嵌入到客服工作台的"推理过程"折叠面板里，主管点开就能看到 AI 的每一步推理和工具调用结果。上线后信任度从 40% 提升到了 85%。
结果与总结
这个 Agent 已经运行了 3 个月，处理了约 5000 次客服会话。一些关键数据：
最大的坑有三个：
延伸思考
这个架构虽然够用，但还有几个方向可以继续优化：
构建 Agent 这件事，说到底不是"让 AI 替人做决定"，而是把 AI 的能力嵌入到人类的工作流中，让 AI 做执行、人类做决策。把握好这个原则，架构就不会走偏。
你可能感兴趣的文章
分类：Android
标签：
评论已关闭！