跳转至

AI Agent 智能体教程

概述

AI Agent(智能体)是能够感知环境、做出决策并执行行动的人工智能系统。通过结合大语言模型(LLM)和工具调用,Agent能够完成复杂的多步骤任务。

Agent核心架构

graph TB
    subgraph 环境
        ENV[外部环境]
    end

    subgraph Agent
        P[感知 Perception]
        D[决策 Decision]
        A[行动 Action]
        M[记忆 Memory]
    end

    subgraph 工具
        T1[搜索]
        T2[代码执行]
        T3[API调用]
        T4[数据库]
    end

    ENV --> P
    P --> D
    D --> A
    A --> ENV

    D --> M
    M --> D

    A --> T1
    A --> T2
    A --> T3
    A --> T4

    style ENV fill:#e3f2fd
    style D fill:#fff3e0
    style A fill:#e8f5e9

ReAct架构流程

flowchart TB
    Q[用户问题] --> T1[Thought: 分析问题]
    T1 --> A1[Action: 选择工具]
    A1 --> O1[Observation: 获取结果]
    O1 --> T2{完成?}
    T2 -->|否| T1
    T2 -->|是| R[返回答案]

    style Q fill:#e3f2fd
    style T1 fill:#fff3e0
    style A1 fill:#e8f5e9
    style R fill:#f3e5f5

Agent vs Chatbot

特性 Chatbot Agent
交互方式 单轮/多轮对话 自主执行任务
能力范围 文本生成 工具调用、任务规划
执行方式 被动响应 主动行动
复杂度 简单问答 多步骤推理

Agent架构模式

1. ReAct(推理+行动)

# ReAct流程

Thought: 分析当前情况,决定下一步行动
Action: 执行具体操作
Observation: 观察执行结果
...重复直到完成...

示例:
Thought: 用户想知道今天的天气,我需要调用天气API
Action: call_weather_api(city="北京")
Observation: 北京今天晴,25°C
Thought: 已获取天气信息,可以回复用户
Action: reply("北京今天天气晴朗,温度25°C,适合出行")

2. Plan-and-Execute

# 规划执行模式

步骤1:制定完整计划
步骤2:按计划逐步执行
步骤3:根据结果调整计划

适用场景:复杂任务、需要全局规划

3. Multi-Agent协作

# 多智能体协作

Agent 1: 研究员 - 收集信息
Agent 2: 分析师 - 分析数据
Agent 3: 作者 - 撰写报告
Agent 4: 审核员 - 质量检查

构建Agent的核心组件

1. LLM(大脑)

# 使用OpenAI API
from openai import OpenAI

client = OpenAI(api_key="your_api_key")

def llm_chat(messages):
    response = client.chat.completions.create(
        model="gpt-5.5",
        messages=messages,
        temperature=0.7
    )
    return response.choices[0].message.content

2. Tools(工具)

# 定义工具
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "搜索互联网获取信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索关键词"
                    }
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "execute_code",
            "description": "执行Python代码",
            "parameters": {
                "type": "object",
                "properties": {
                    "code": {
                        "type": "string",
                        "description": "Python代码"
                    }
                },
                "required": ["code"]
            }
        }
    }
]

3. Memory(记忆)

class AgentMemory:
    def __init__(self):
        self.short_term = []  # 短期记忆(当前对话)
        self.long_term = []   # 长期记忆(持久化)

    def add_message(self, role, content):
        self.short_term.append({"role": role, "content": content})

    def get_context(self, max_messages=10):
        return self.short_term[-max_messages:]

    def save_to_long_term(self, info):
        self.long_term.append(info)

4. Planning(规划)

class Planner:
    def create_plan(self, task):
        prompt = f"""
        请为以下任务制定执行计划:

        任务:{task}

        请输出:
        1. 目标分解
        2. 执行步骤
        3. 所需资源
        4. 预期时间
        """
        return llm_chat([{"role": "user", "content": prompt}])

Agent实现示例

简单ReAct Agent

import json
from openai import OpenAI

class ReActAgent:
    def __init__(self, api_key):
        self.client = OpenAI(api_key=api_key)
        self.tools = {}
        self.memory = []

    def register_tool(self, name, func, description):
        self.tools[name] = {"func": func, "description": description}

    def run(self, task, max_iterations=10):
        messages = [
            {"role": "system", "content": self._get_system_prompt()},
            {"role": "user", "content": task}
        ]

        for i in range(max_iterations):
            response = self.client.chat.completions.create(
                model="gpt-5.5",
                messages=messages,
                tools=self._get_tools_schema(),
                tool_choice="auto"
            )

            choice = response.choices[0]

            if choice.message.tool_calls:
                # 执行工具调用
                tool_call = choice.message.tool_calls[0]
                result = self._execute_tool(tool_call)

                messages.append(choice.message)
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": str(result)
                })
            else:
                # 最终回复
                return choice.message.content

        return "达到最大迭代次数,任务未完成"

    def _get_system_prompt(self):
        tools_desc = "\n".join([
            f"- {name}: {info['description']}" 
            for name, info in self.tools.items()
        ])

        return f"""你是一个智能助手,可以使用以下工具:

{tools_desc}

请按照以下格式思考和行动:
Thought: 分析当前情况
Action: 选择要执行的工具
Observation: 观察工具执行结果
...重复直到完成任务

当任务完成时,直接输出最终答案。"""

    def _get_tools_schema(self):
        # 构建OpenAI tools格式
        tools_schema = []
        for name, info in self.tools.items():
            tools_schema.append({
                "type": "function",
                "function": {
                    "name": name,
                    "description": info["description"],
                    "parameters": {"type": "object", "properties": {}}
                }
            })
        return tools_schema

    def _execute_tool(self, tool_call):
        func_name = tool_call.function.name
        args = json.loads(tool_call.function.arguments)
        return self.tools[func_name]["func"](**args)

使用示例

# 创建Agent
agent = ReActAgent(api_key="your_api_key")

# 注册工具
agent.register_tool(
    name="search",
    func=lambda query: f"搜索结果:关于{query}的信息...",
    description="搜索互联网获取信息"
)

agent.register_tool(
    name="calculate",
    func=lambda expression: eval(expression),
    description="执行数学计算"
)

# 运行Agent
result = agent.run("搜索今天的天气,然后计算25度转换成华氏度是多少")
print(result)

LangChain框架

基础使用

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool

# 初始化LLM
llm = ChatOpenAI(model="gpt-5.5", api_key="your_api_key")

# 定义工具
@tool
def search(query: str) -> str:
    """搜索互联网获取信息"""
    return f"搜索结果:{query}的相关信息..."

@tool
def calculator(expression: str) -> str:
    """执行数学计算"""
    return str(eval(expression))

# 创建Prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有用的助手。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

# 创建Agent
tools = [search, calculator]
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 运行
result = executor.invoke({"input": "今天北京天气如何?25度是多少华氏度?"})
print(result["output"])

使用LangGraph构建复杂Agent

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

# 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    next_step: str

# 定义节点
def research_node(state):
    # 研究节点逻辑
    return {"messages": [research_result]}

def analysis_node(state):
    # 分析节点逻辑
    return {"messages": [analysis_result]}

def write_node(state):
    # 写作节点逻辑
    return {"messages": [final_report]}

# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("research", research_node)
workflow.add_node("analysis", analysis_node)
workflow.add_node("write", write_node)

workflow.set_entry_point("research")
workflow.add_edge("research", "analysis")
workflow.add_edge("analysis", "write")
workflow.add_edge("write", END)

# 编译运行
app = workflow.compile()
result = app.invoke({"messages": [], "next_step": "research"})

Agent应用场景

1. 代码助手Agent

# 功能
- 代码生成
- Bug调试
- 代码审查
- 文档生成

# 实现思路
1. 接收用户代码需求
2. 分析需求,生成代码
3. 执行测试验证
4. 返回结果

2. 数据分析Agent

# 功能
- 数据清洗
- 统计分析
- 可视化生成
- 报告撰写

# 工具集
- pandas:数据处理
- matplotlib/seaborn:可视化
- jupyter:交互式执行

3. 研究助手Agent

# 功能
- 文献检索
- 信息提取
- 总结归纳
- 引用管理

# 工具集
- arxiv API:论文搜索
- Semantic Scholar:引用查询
- Zotero:文献管理

4. 客服Agent

# 功能
- 问题理解
- 知识检索
- 回答生成
- 工单创建

# 特点
- 多轮对话
- 情感识别
- 无缝转人工

Agent开发最佳实践

1. 工具设计原则

# 工具应该:

✅ 单一职责:每个工具只做一件事
✅ 清晰描述:让LLM理解何时使用
✅ 错误处理:优雅地处理异常
✅ 返回格式:返回结构化数据

2. 错误处理

try:
    result = tool.execute(args)
except ToolError as e:
    result = f"工具执行失败:{str(e)},请尝试其他方法"
except Exception as e:
    result = f"未知错误:{str(e)},请重试"

3. 安全考虑

# 安全原则

1. 输入验证:验证所有用户输入
2. 权限控制:限制工具访问权限
3. 沙箱执行:在隔离环境运行代码
4. 日志记录:记录所有操作
5. 速率限制:防止滥用

4. 性能优化

# 优化策略

1. 缓存:缓存常用结果
2. 并行:并行执行独立任务
3. 懒加载:按需加载工具
4. 简化prompt:减少token使用

常见问题

Q1: Agent总是陷入循环怎么办?

# 解决方案
1. 设置最大迭代次数
2. 添加循环检测
3. 使用plan-and-execute模式

Q2: Agent不调用工具怎么办?

# 优化prompt

明确说明何时使用工具:
- 当需要最新信息时,使用search工具
- 当需要计算时,使用calculator工具

提供工具使用示例

Q3: 如何调试Agent?

# 调试技巧

1. 开启verbose模式
2. 记录所有LLM调用
3. 打印工具输入输出
4. 使用断点调试

学习资源

# 框架文档
- LangChain:https://python.langchain.com
- LangGraph:https://langchain-ai.github.io/langgraph/
- AutoGPT:https://github.com/Significant-Gravitas/AutoGPT

# 参考资料
- LLM Agent论文
- Tool Learning论文
- ReAct论文

下一步学习

  1. Prompt Engineering教程 - 优化Agent的prompt
  2. [LangChain框架实战] - 深入学习Agent框架
  3. [多Agent协作] - 构建复杂Agent系统