Agent基础架构

Agent（智能体）是能够自主规划、执行任务、使用工具的AI系统。本节介绍Agent的核心架构和实现原理。

什么是Agent？

Agent vs 普通程序

普通程序：
输入 → 固定逻辑 → 输出

Agent：
输入 → 思考 → 决策 → 行动 → 观察 → 再思考 → ... → 输出
       ↑_________________循环__________________↓

Agent的核心能力

能力	说明
感知	理解用户输入和环境信息
规划	分解任务，制定执行步骤
行动	调用工具执行操作
反思	评估结果，调整策略
记忆	存储和调用历史信息

Agent架构

ReAct架构

最流行的Agent架构：Reasoning + Acting

用户输入
    ↓
┌─────────────────────────────────────┐
│            Agent 循环               │
│  ┌───────────┐    ┌───────────┐    │
│  │   思考    │ → │   行动    │    │
│  │ (Thought) │    │ (Action)  │    │
│  └───────────┘    └─────┬─────┘    │
│        ↑                ↓          │
│  ┌───────────┐    ┌───────────┐    │
│  │   决策    │ ← │   观察    │    │
│  │           │    │(Observation)│  │
│  └───────────┘    └───────────┘    │
└─────────────────────────────────────┘
    ↓
最终答案

执行流程示例

用户：北京今天天气怎么样？温度加上10度是多少？

循环1:
Thought: 用户想知道北京天气，我需要调用天气工具
Action: get_weather("北京")
Observation: 北京今天晴天，气温25°C

循环2:
Thought: 我获得了天气信息，现在需要计算温度+10
Action: calculate("25 + 10")
Observation: 35

循环3:
Thought: 我已经获得所有需要的信息，可以回答用户了
Final Answer: 北京今天晴天，气温25°C。加上10度后是35°C。

使用LangGraph构建Agent

LangGraph是构建Agent的新框架，比LangChain更灵活：

安装

pip install langgraph

基本Agent

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

# 定义工具
@tool
def search(query: str) -> str:
    """搜索信息"""
    return f"搜索结果: {query}"

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        return str(eval(expression))
    except:
        return "计算错误"

tools = [search, calculate]

# 定义状态
class AgentState(TypedDict):
    messages: list

# 创建模型和工具节点
model = ChatOpenAI(model="gpt-4o").bind_tools(tools)
tool_node = ToolNode(tools)

# 定义节点函数
def agent(state: AgentState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

# 判断是否需要调用工具
def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent)
workflow.add_node("tools", tool_node)

workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("tools", "agent")

# 编译运行
app = workflow.compile()

# 使用
result = app.invoke({
    "messages": [{"role": "user", "content": "北京天气怎么样？"}]
})

Agent的核心组件

1. 状态管理

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    # 使用add_messages自动管理消息历史
    messages: Annotated[list, add_messages]
    # 自定义状态
    current_step: int
    max_steps: int

2. 工具定义

from langchain_core.tools import tool
import requests

@tool
def web_search(query: str) -> str:
    """搜索互联网信息
    
    Args:
        query: 搜索关键词
        
    Returns:
        搜索结果摘要
    """
    # 实际调用搜索API
    response = requests.get(f"https://api.example.com/search?q={query}")
    return response.json().get("summary", "无结果")

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件
    
    Args:
        to: 收件人邮箱
        subject: 邮件主题  
        body: 邮件正文
    """
    # 发送逻辑
    return f"邮件已发送至 {to}"

3. 提示词设计

from langchain_core.prompts import ChatPromptTemplate

system_prompt = """你是一个智能助手，可以使用工具帮助用户完成任务。

你可以使用以下工具：
{tool_names}

使用工具时请遵循以下格式：
- Thought: 思考下一步做什么
- Action: 选择要使用的工具
- Action Input: 工具的输入参数

完成所有任务后，使用Final Answer给出最终答案。
"""

prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    ("placeholder", "{chat_history}"),
    ("user", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

Agent类型

1. ReAct Agent

推理+行动模式，最通用：

from langchain.agents import create_react_agent

agent = create_react_agent(model, tools, prompt)

2. 工具调用Agent

利用模型的function calling能力：

from langchain.agents import create_tool_calling_agent

agent = create_tool_calling_agent(model, tools, prompt)

3. Plan-and-Execute Agent

先规划再执行：

# 第一步：规划
plan = model.invoke("制定计划来完成：{task}")

# 第二步：逐步执行
for step in plan.steps:
    result = execute_step(step)

防止无限循环

Agent可能陷入死循环，需要设置限制：

from langgraph.checkpoint.memory import MemorySaver

# 设置递归限制
app = workflow.compile(
    checkpointer=MemorySaver()
)

config = {"configurable": {"thread_id": "test"}, "recursion_limit": 10}

result = app.invoke(input_data, config)

小结

概念	说明
Agent	自主规划和执行的AI系统
ReAct	推理+行动的架构模式
状态	Agent执行过程中的数据
工具	Agent可调用的能力
循环限制	防止无限循环

下一步

继续学习 工具调用与规划，深入了解Agent的高级能力。