乐于分享
好东西不私藏

AI Agents 从零上手(Generative AI for Beginners第08篇)

AI Agents 从零上手(Generative AI for Beginners第08篇)

# AI Agents 从零上手(第08篇)

这是微软开源课程《Generative AI for Beginners》的中文精讲系列,第8篇,对应原课第17课:AI Agents(智能代理)的核心原理、设计模式与完整实现。这是当下 AI 领域最热、最有潜力的方向。

什么是 AI Agent?

聊天机器人:你问一句,它答一句,结束。

AI Agent:你给一个目标,它自己规划步骤,调用各种工具,迭代执行,直到完成任务。 `

普通 LLM:问题 → 答案(一次交互)

AI Agent:目标 → [思考→行动→观察] 循环 → 最终结果(多步骤自主执行)

`

这个"思考→行动→观察"的循环,就是 Agent 的核心——叫做 ReAct 模式(Reasoning + Acting)。


一、Agent 的核心架构

一个完整的 AI Agent 由四个部分组成:

组件
职责
大脑(LLM)
推理、规划、决策
工具(Tools)
执行具体动作(搜索、计算、API调用)
记忆(Memory)
存储短期对话历史和长期经验
规划器(Planner)
把复杂目标分解成可执行步骤

二、从零实现 ReAct Agent

2.1 基础 ReAct 实现

`python 

from openai import OpenAI

import json

import re

client = OpenAI()

# 定义工具集

def web_search(query: str) -> str:

"""模拟网络搜索"""

# 真实场景:调用 Bing/Google Search API

mock_results = {

"Python最新版本": "Python 3.13 于2024年10月发布,新增类型推断和JIT编译器",

"OpenAI GPT-4o": "GPT-4o 是 OpenAI 最新的多模态模型,支持文字、图像、音频",

"北京天气": "北京今日晴,气温22-30°C,紫外线强,建议防晒"

}

for key in mock_results:

if key in query:

return mock_results[key]

return f"搜索结果:关于'{query}',暂未找到具体信息"

def calculator(expression: str) -> str:

"""数学计算器"""

import math

try:

safe_globals = {k: getattr(math, k) for k in dir(math) if not k.startswith('_')}

result = eval(expression, {"__builtins__": {}}, safe_globals)

return str(result)

except Exception as e:

return f"计算错误: {e}"

def get_current_time() -> str:

"""获取当前时间"""

from datetime import datetime

return datetime.now().strftime("%Y年%m月%d日 %H:%M:%S")

def read_file(filename: str) -> str:

"""读取文件内容"""

try:

with open(filename, 'r', encoding='utf-8') as f:

return f.read()[:2000]  # 限制返回长度

except Exception as e:

return f"读取失败: {e}"

# 工具注册表

TOOLS = {

"web_search": {

"fn": web_search,

"description": "搜索网络上的最新信息",

"params": "query: str (搜索关键词)"

},

"calculator": {

"fn": calculator,

"description": "执行数学计算,支持 Python 数学表达式",

"params": "expression: str (数学表达式,如 '2**10' 或 'sqrt(144)')"

},

"get_current_time": {

"fn": get_current_time,

"description": "获取当前日期和时间",

"params": "无参数"

},

"read_file": {

"fn": read_file,

"description": "读取本地文件内容",

"params": "filename: str (文件路径)"

}

}

`

2.2 ReAct 提示词模板

`python 

REACT_SYSTEM_PROMPT = """你是一个智能助手,可以通过"思考-行动-观察"的循环来解决复杂问题。

可用工具:

{tools_desc}

回答格式(严格遵守):

思考: [分析当前情况,决定下一步]

行动: [工具名称]

参数: [工具参数,JSON格式]

或者当你有足够信息时:

思考: [已获得足够信息]

最终答案: [给用户的最终回答]

规则:

  • 每次只执行一个行动
  • 必须基于观察结果来决定下一步
  • 如果工具返回错误,换一种方式尝试
  • 达到目标后立即给出最终答案
""" 

def format_tools_desc() -> str:

lines = []

for name, info in TOOLS.items():

lines.append(f"- {name}: {info['description']}")

lines.append(f"  参数: {info['params']}")

return "\n".join(lines)

`

2.3 完整 Agent 执行循环

`python 

class ReActAgent:

"""ReAct 模式 AI Agent"""

def __init__(self, max_steps: int = 10):

self.max_steps = max_steps

self.system_prompt = REACT_SYSTEM_PROMPT.format(

tools_desc=format_tools_desc()

)

def _parse_action(self, text: str) -> tuple[str, dict]:

"""解析 LLM 输出中的行动和参数"""

action_match = re.search(r'行动:\s*(\w+)', text)

params_match = re.search(r'参数:\s(\{.?\})', text, re.DOTALL)

if not action_match:

return None, {}

action = action_match.group(1).strip()

params = {}

if params_match:

try:

params = json.loads(params_match.group(1))

except:

# 尝试提取简单字符串参数

text_match = re.search(r'参数:\s*["\']?([^"\'\n{]+)', text)

if text_match:

params = {"query": text_match.group(1).strip()}

return action, params

def _execute_tool(self, action: str, params: dict) -> str:

"""执行工具并返回结果"""

if action not in TOOLS:

return f"错误:未知工具 '{action}',可用工具: {list(TOOLS.keys())}"

tool_fn = TOOLS[action]["fn"]

try:

if params:

result = tool_fn(**params)

else:

result = tool_fn()

return str(result)

except Exception as e:

return f"工具执行错误: {e}"

def run(self, task: str, verbose: bool = True) -> str:

"""运行 Agent 完成任务"""

messages = [

{"role": "system", "content": self.system_prompt},

{"role": "user", "content": task}

]

if verbose:

print(f"\n{'='*60}")

print(f"任务: {task}")

print('='*60)

for step in range(self.max_steps):

# LLM 决策

response = client.chat.completions.create(

model="gpt-4o",

messages=messages,

temperature=0.1  # Agent 决策用低温度保证稳定性

)

llm_output = response.choices[0].message.content

messages.append({"role": "assistant", "content": llm_output})

if verbose:

print(f"\n[步骤 {step+1}]")

print(llm_output)

# 检查是否有最终答案

if "最终答案:" in llm_output:

final = re.search(r'最终答案:\s*(.+)', llm_output, re.DOTALL)

if final:

return final.group(1).strip()

# 解析并执行行动

action, params = self._parse_action(llm_output)

if not action:

# LLM 没有给出行动,可能是在等待

if step == self.max_steps - 1:

return llm_output

continue

# 执行工具

observation = self._execute_tool(action, params)

if verbose:

print(f"\n[工具结果] {observation}")

# 把观察结果加入历史

messages.append({

"role": "user",

"content": f"观察结果: {observation}\n\n继续执行下一步:"

})

return "已达到最大步骤数,任务未完成"

# 使用示例

agent = ReActAgent(max_steps=8)

print(agent.run("计算 2024 年到现在过了多少天?今天是哪一天?"))

print(agent.run("搜索 Python 的最新版本,然后告诉我它的主要新特性"))

print(agent.run("计算 100 的平方根,以及 2 的 10 次方分别是多少?两个数相加等于多少?"))

`

三、多 Agent 系统

复杂任务可以拆分给多个专业 Agent 协作完成:

`python 

class MultiAgentOrchestrator:

"""多 Agent 协作框架"""

def __init__(self):

self.agents = {}

self._register_default_agents()

def _register_default_agents(self):

"""注册默认 Agent"""

self.register("researcher", "你是研究员,擅长信息搜集和整理")

self.register("analyst", "你是分析师,擅长数据分析和逻辑推理")

self.register("writer", "你是写作专家,擅长将信息整合成清晰的文章")

def register(self, name: str, role: str):

self.agents[name] = role

def delegate(self, agent_name: str, task: str) -> str:

"""委派任务给指定 Agent"""

if agent_name not in self.agents:

return f"未知 Agent: {agent_name}"

response = client.chat.completions.create(

model="gpt-4o",

messages=[

{"role": "system", "content": self.agents[agent_name]},

{"role": "user", "content": task}

]

)

return response.choices[0].message.content

def pipeline(self, goal: str) -> str:

"""执行多 Agent 流水线"""

# 1. 研究员收集信息

research_result = self.delegate(

"researcher",

f"围绕以下目标,收集和整理关键信息:{goal}"

)

# 2. 分析师分析

analysis_result = self.delegate(

"analyst",

f"基于以下研究结果进行深度分析:\n{research_result}\n\n原始目标:{goal}"

)

# 3. 写手整合

final_result = self.delegate(

"writer",

f"将以下分析整合成一份清晰的报告:\n{analysis_result}"

)

return final_result

orchestrator = MultiAgentOrchestrator()

report = orchestrator.pipeline("分析生成式AI对软件开发行业的影响")

print(report)

`

四、Agent 设计最佳实践

问题
解决方案
无限循环
设置最大步骤数,超出则中断
工具调用失败
错误信息返回给 LLM,让它换方式重试
上下文过长
对 Agent 历史做滚动摘要
不可预期行为
关键操作加二次确认("确认要执行吗?")
安全边界
明确禁止的操作(删除文件、发送邮件)需人工授权

小结

  • ✅ Agent 核心:LLM + 工具 + 记忆 + 规划 = 能自主完成任务的 AI
  • ✅ ReAct 模式:思考→行动→观察循环,让 AI 逐步解决复杂问题
  • ✅ 工具扩展:搜索、计算、文件操作……任何 Python 函数都能成为工具
  • ✅ 多 Agent:将复杂任务分解给专业 Agent,流水线协作完成

下一篇预告

下一篇我们进入Fine-Tuning(微调)与小语言模型(SLM)——当 Prompt 工程和 RAG 还不够用,微调才是终极武器。我们会讲清楚什么时候该微调、怎么做,以及 SLM 的崛起与应用场景。

关注不迷路,下期见。
本文基于微软开源课程《Generative AI for Beginners》第17课整理,原课程地址:https://github.com/microsoft/generative-ai-for-beginners