AI应用开发系列(四) AI-Agent开发-让大模型动起来

AI Agent 开发：让大模型"动起来"——从"会聊天"到"会干活"

系列导读：这是「 AI 应用开发」第 4 篇。前三篇咱们聊了全景图、模型接入、RAG 知识库。今天进入最激动人心的部分：怎么让 AI 不仅能回答问题，还能真正"干活"——查数据库、调 API、写报告、发邮件？这就是 AI Agent 要解决的问题。

一、问题引入：AI 只会说，不会动

你的 RAG 知识库搭好了，员工问"咱们公司年假政策"，AI 能准确回答。但有人问了：

“帮我查一下我今年的年假还剩几天，然后发封邮件给 HR 申请下周休 3 天。”

AI 一脸懵逼：“根据《员工手册》，年假申请需要提前 3 天提交…” 它只会复读文档，没法查你的实际剩余年假，也没法真的发邮件。

核心矛盾：大模型是"大脑"，但企业应用需要"手脚"——跟外部系统交互的能力。

朴素的解决方案：在代码里写死逻辑？

# 别这么干！这是传统的 if-else 机器人defhandle_request(user_input):if"查年假"inuser_input:days=query_hr_system(user_id)returnf"您还剩{days}天年假"elif"发邮件"inuser_input:send_email(to,subject,content)return"邮件已发送"# ... 每加一个功能就要改这里

但这有几个致命问题：

1. 意图识别脆弱：用户说"我想休息几天"，你识别不出来
2. 组合场景搞不定：“查年假 + 发邮件” 这种多步骤任务，代码爆炸
3. 维护成本高：每加一个功能，就要改核心逻辑
4. 不灵活：用户换个说法，系统就不认识了

问题来了：怎么让 AI 自己理解要做什么、自己决定调用哪些工具、自己处理多步骤任务？

二、方案分析：Agent 的核心思路

Agent（智能体）的思路是：给大模型"工具"和"自主权"，让它自己规划、自己执行。

就像你给实习生一份任务清单和一套工具，告诉他：“你去把这个报告做出来，需要查数据就用数据库工具，需要画图就用图表工具，遇到问题来问我。”

Agent 的核心循环（ReAct 模式）：

┌─────────────────────────────────────────┐ │ 1. 思考（Thought）：分析当前状态，决定下一步 │ │ "用户要查年假，我需要调用 HR 系统" │ ├─────────────────────────────────────────┤ │ 2. 行动（Action）：调用工具 │ │ 调用 query_leave_days(user_id) │ ├─────────────────────────────────────────┤ │ 3. 观察（Observation）：获取工具返回结果 │ │ "剩余年假：5 天" │ ├─────────────────────────────────────────┤ │ 4. 循环：回到步骤 1，直到任务完成 │ │ "用户还要发邮件，我需要调用邮件工具..." │ └─────────────────────────────────────────┘

Agent 需要的"装备"：

1. 工具（Tools）：可调用的函数/API，如查数据库、发邮件、搜索网页
2. 记忆（Memory）：记住之前的对话和中间结果
3. 规划（Planning）：拆解复杂任务，决定执行顺序
4. 大模型（LLM）：作为"大脑"，做决策和推理

三、实现过程：Step by Step

第一步：定义工具接口

工具是 Agent 的"手脚"。统一接口，让 Agent 能动态发现和调用。

@dataclassclassTool:"""工具定义：描述 + 执行函数"""name:str# 工具名description:str# 功能描述（给 AI 看的！很重要）parameters:Dict# JSON Schema 参数定义func:Callable# 实际执行函数@dataclassclassToolResult:"""工具执行结果"""success:booldata:Any error:Optional[str]=None

几个实际的企业工具示例：

# 工具 1：查询 HR 系统（年假余额）defquery_leave_balance(employee_id:str)->ToolResult:"""查询员工年假余额"""try:# 实际调用 HR 系统 APIresult=hr_api.get(f"/employees/{employee_id}/leave-balance")returnToolResult(success=True,data=result.json())exceptExceptionase:returnToolResult(success=False,error=str(e))leave_tool=Tool(name="query_leave_balance",description="查询指定员工的年假余额。参数 employee_id 是员工工号。",parameters={"type":"object","properties":{"employee_id":{"type":"string","description":"员工工号，如 EMP001"}},"required":["employee_id"]},func=query_leave_balance)# 工具 2：发送邮件defsend_email(to:str,subject:str,body:str)->ToolResult:"""发送邮件"""try:email_service.send(to=to,subject=subject,body=body)returnToolResult(success=True,data={"message_id":"xxx"})exceptExceptionase:returnToolResult(success=False,error=str(e))email_tool=Tool(name="send_email",description="发送邮件给指定收件人。需要提供收件人邮箱、主题和正文。",parameters={"type":"object","properties":{"to":{"type":"string","description":"收件人邮箱"},"subject":{"type":"string","description":"邮件主题"},"body":{"type":"string","description":"邮件正文"}},"required":["to","subject","body"]},func=send_email)# 工具 3：查询数据库（通用 SQL）defquery_database(sql:str)->ToolResult:"""执行 SQL 查询（只读）"""try:# 安全校验：只允许 SELECTifnotsql.strip().upper().startswith("SELECT"):returnToolResult(success=False,error="只允许执行 SELECT 查询")result=db.execute(sql)returnToolResult(success=True,data=result.fetchall())exceptExceptionase:returnToolResult(success=False,error=str(e))db_tool=Tool(name="query_database",description="执行数据库查询。只支持 SELECT 语句。参数 sql 是 SQL 查询语句。",parameters={"type":"object","properties":{"sql":{"type":"string","description":"SQL 查询语句，如 SELECT * FROM orders WHERE id = 123"}},"required":["sql"]},func=query_database)

关键设计：

• description是给 AI 看的"说明书"，写得越清楚，AI 调用越准
• parameters用 JSON Schema，让 AI 知道要传什么参数
• 安全校验内置在工具里，不让 AI 直接操作危险功能

第二步：ReAct Agent 实现

ReAct（Reasoning + Acting）是最经典的 Agent 模式。核心思想：让 AI 在"思考"和"行动"之间循环。

classReActAgent:"""ReAct 模式 Agent：思考 → 行动 → 观察 → 循环"""def__init__(self,llm:LLMProvider,tools:List[Tool],max_iterations:int=10):self.llm=llm self.tools={t.name:tfortintools}self.max_iterations=max_iterationsdefrun(self,user_input:str,context:Dict=None)->str:"""执行用户请求"""# 初始化记忆memory=AgentMemory()memory.add(Message(role="user",content=user_input))# 构建系统提示system_prompt=self._build_system_prompt()foriinrange(self.max_iterations):# 1. 让 AI 思考下一步response=self.llm.chat(messages=[Message(role="system",content=system_prompt),*memory.get_history()],config=ChatConfig(temperature=0.2)# Agent 需要确定性，温度要低)content=response.content memory.add(Message(role="assistant",content=content))# 2. 解析 AI 的决策action=self._parse_action(content)ifaction.type=="finish":# 任务完成，返回结果returnaction.resultelifaction.type=="tool_call":# 3. 执行工具tool=self.tools.get(action.tool_name)ifnottool:observation=f"错误：工具 '{action.tool_name}' 不存在"else:result=tool.func(**action.parameters)observation=self._format_observation(result)# 4. 把观察结果加入记忆memory.add(Message(role="user",content=f"观察结果：{observation}"))else:# AI 在纯思考，继续循环passreturn"达到最大迭代次数，任务未完成"def_build_system_prompt(self)->str:"""构建系统提示，告诉 AI 怎么思考、有哪些工具"""tools_desc="\n".join([f"-{name}:{tool.description}\n 参数：{json.dumps(tool.parameters,ensure_ascii=False)}"forname,toolinself.tools.items()])returnf"""你是一个智能助手，可以通过调用工具来完成用户请求。 可用工具：{tools_desc}工作方式：1.分析用户需求，决定是否需要调用工具2.如果需要，按以下格式输出工具调用： ```action{{"tool":"工具名","parameters":{{"参数名":"参数值"}}}}

3. 观察工具返回结果，决定下一步
4. 任务完成后，输出：

   {{"type": "finish", "result": "最终回答"}}

注意事项：

• 每次只调用一个工具
• 如果工具返回错误，分析原因并尝试修复
• 不要编造工具返回的内容
  “”"

### 第三步：看一个完整的执行过程 用户说："帮我查一下年假还剩几天，然后发邮件给 hr@company.com 申请下周休 3 天。" **Round 1 - AI 思考：**

用户有两个需求：

1. 查询年假余额
2. 发邮件申请休假

我需要先查询年假余额，然后发邮件。

先调用 query_leave_balance。

**Round 1 - AI 输出：** ```action { "tool": "query_leave_balance", "parameters": {"employee_id": "EMP001"} }

Round 1 - 执行工具：

观察结果：{"annual_leave": 5, "sick_leave": 3, "personal_leave": 2}

Round 2 - AI 思考：

年假还剩 5 天，足够休 3 天。现在需要发邮件申请。 邮件内容应该包含：申请 3 天年假、具体时间（下周）。

Round 2 - AI 输出：

{ "tool": "send_email", "parameters": { "to": "hr@company.com", "subject": "年假申请 - EMP001", "body": "HR 同事你好，\n\n我计划下周申请 3 天年假（具体日期待定），目前剩余年假 5 天，请审批。\n\n谢谢！" } }

Round 2 - 执行工具：

观察结果：{"message_id": "msg_abc123", "status": "sent"}

Round 3 - AI 思考：

两个任务都完成了： 1. 查询到年假余额：5 天 2. 邮件已发送给 HR 可以结束任务了。

Round 3 - AI 输出：

{ "type": "finish", "result": "已为您完成以下操作：\n1. 查询年假余额：您目前还剩 5 天年假\n2. 发送申请邮件：已向 hr@company.com 发送年假申请邮件（申请 3 天）\n\nHR 审批通过后，您会收到通知。" }

任务完成！AI 自主完成了多步骤任务，中间没有人工干预。

第四步：Function Calling 模式（更简洁的实现）

上面是 ReAct 的"文本解析"模式，实际生产环境更推荐用模型的原生 Function Calling：

classFunctionCallingAgent:"""基于模型原生 Function Calling 的 Agent"""defrun(self,user_input:str)->str:messages=[Message(role="user",content=user_input)]foriinrange(self.max_iterations):# 直接让模型决定调用哪个工具response=self.llm.function_call(messages=messages,functions=[t.parametersfortinself.tools.values()])ifresponse.finish_reason=="function_call":# 模型决定调用工具tool_name=response.function_call.name arguments=json.loads(response.function_call.arguments)# 执行工具tool=self.tools[tool_name]result=tool.func(**arguments)# 把结果加回对话messages.append(Message(role="function",name=tool_name,content=json.dumps(result.dataifresult.successelseresult.error)))else:# 模型直接回答了returnresponse.contentreturn"达到最大迭代次数"

Function Calling vs ReAct 文本模式：

第五步：记忆管理——Agent 不能"健忘"

Agent 需要记住之前说过的话、做过的事。

classAgentMemory:"""Agent 记忆管理"""def__init__(self,max_messages:int=20):self.messages=[]self.max_messages=max_messages self.summary=""# 长对话的摘要defadd(self,message:Message):self.messages.append(message)# 超过长度限制，做摘要iflen(self.messages)>self.max_messages:self._summarize()def_summarize(self):"""用 LLM 把早期对话摘要"""early_messages=self.messages[:10]prompt=f"请摘要以下对话的关键信息：\n{format_messages(early_messages)}"summary=self.llm.chat([Message(role="user",content=prompt)]).content self.summary=summary self.messages=self.messages[10:]defget_history(self)->List[Message]:"""获取完整历史（摘要 + 近期消息）"""ifself.summary:return[Message(role="system",content=f"历史摘要：{self.summary}")]+self.messagesreturnself.messages

记忆的类型：

• 短期记忆：当前对话的上下文
• 长期记忆：用户偏好、历史交互摘要（存数据库）
• 实体记忆：提取的关键信息（如用户工号、常用查询）

四、进阶：Multi-Agent 协作

复杂任务一个 Agent 搞不定，需要多个 Agent 协作。

classMultiAgentSystem:"""多 Agent 协作系统"""def__init__(self):# 定义多个专业 Agentself.agents={"planner":Agent(name="规划师",role="分析用户需求，制定执行计划",tools=[]),"researcher":Agent(name="研究员",role="查询数据、搜索信息",tools=[db_tool,search_tool]),"writer":Agent(name="写手",role="撰写报告、邮件",tools=[email_tool]),"reviewer":Agent(name="审核员",role="检查内容准确性和合规性",tools=[])}defexecute(self,task:str)->str:# 1. 规划师制定计划plan=self.agents["planner"].run(f"请为以下任务制定执行计划：{task}")# 2. 按步骤分配给其他 Agentresults=[]forstepinparse_plan(plan):ifstep.type=="research":result=self.agents["researcher"].run(step.description)elifstep.type=="write":result=self.agents["writer"].run(step.description)results.append(result)# 3. 审核员检查final=self.agents["reviewer"].run(f"请审核以下内容：{combine_results(results)}")returnfinal

实际场景：

• 数据分析报告：研究员查数据 → 写手生成图表和文字 → 审核员检查
• 客服工单处理：理解 Agent 分析意图 → 查询 Agent 查订单 → 操作 Agent 执行退款 → 通知 Agent 发邮件

五、小结

今天咱们实现了一套完整的 AI Agent 系统：

Agent 的核心价值：让 AI 从"问答机器人"变成"数字员工"——能理解意图、能调用工具、能完成多步骤任务。

安全提醒：

• 工具权限要最小化（只读查询 vs 写操作）
• 敏感操作要人工确认（如转账、删除数据）
• 所有工具调用要记录审计日志

你在设计 Agent 时最头疼的是什么？是工具描述的写法、还是多步骤任务的稳定性？欢迎交流！