第七章：Agent 代理系统（最新 API）

重要更新：LangChain 最新版本引入了全新的 Agent API。initialize_agent、AgentExecutor 等旧 API 已废弃，请使用本章介绍的 create_agent 新 API。

什么是 Agent？

Agent（代理）是让 LLM 自主决定使用哪些工具来完成任务的系统：

1. 接收任务：用户提出问题
2. 思考决策：LLM 分析需要使用什么工具
3. 执行工具：调用工具获取结果
4. 继续思考：根据结果决定下一步
5. 输出答案：给出最终回复

新旧 API 对比

废弃 API	新 API	说明
initialize_agent()	create_agent()	统一的 agent 创建函数
AgentExecutor	create_agent() 返回的 Runnable	内置 LangGraph 图执行
load_tools()	@tool 装饰器	更灵活的工具定义
ConversationBufferMemory	checkpointer + state	基于图的状态管理

定义工具

@tool 装饰器（推荐）

from langchain.tools import tool

@tool
def get_weather(city: str) -> str:
    """获取城市天气信息。

    Args:
        city: 城市名称，如"北京"、"上海"
    """
    weather_data = {
        "北京": "☀️ 晴天，25°C",
        "上海": "⛅ 多云，22°C",
        "广州": "🌧️ 小雨，28°C"
    }
    return weather_data.get(city, f"未找到 {city} 的天气信息")

@tool
def calculate(expression: str) -> str:
    """计算数学表达式。

    Args:
        expression: 数学表达式，如 "2+2"、"10*5"
    """
    try:
        result = eval(expression)
        return f"计算结果: {result}"
    except Exception as e:
        return f"计算错误: {e}"

复杂参数（Pydantic）

from pydantic import BaseModel, Field
from typing import Literal

class WeatherInput(BaseModel):
    """天气查询参数"""
    location: str = Field(description="城市名称或坐标")
    units: Literal["celsius", "fahrenheit"] = Field(
        default="celsius",
        description="温度单位"
    )
    include_forecast: bool = Field(
        default=False,
        description="是否包含未来5天预报"
    )

@tool(args_schema=WeatherInput)
def get_weather_advanced(
    location: str, 
    units: str = "celsius", 
    include_forecast: bool = False
) -> str:
    """获取详细天气信息"""
    temp = 25 if units == "celsius" else 77
    result = f"{location} 当前温度: {temp}°{units[0].upper()}"
    if include_forecast:
        result += "\n未来5天: 晴转多云"
    return result

访问运行时上下文

from langchain.tools import tool, ToolRuntime
from langgraph.types import Command

@tool
def get_user_info(runtime: ToolRuntime) -> str:
    """获取当前用户信息"""
    messages = runtime.state["messages"]
    return f"当前对话有 {len(messages)} 条消息"

@tool
def set_preference(key: str, value: str, runtime: ToolRuntime) -> Command:
    """设置用户偏好"""
    return Command(update={"preferences": {key: value}})

创建 Agent

最简单的 Agent

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4.1")

agent = create_agent(
    model=model,
    system_prompt="你是一个有帮助的助手。"
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "你好！"}]
})

print(result["messages"][-1].content)

带工具的 Agent

from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI

@tool
def get_weather(city: str) -> str:
    """获取城市天气信息"""
    weather_data = {
        "北京": "晴天，25°C",
        "上海": "多云，22°C",
    }
    return weather_data.get(city, f"未找到 {city} 的天气")

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        return f"结果: {eval(expression)}"
    except Exception as e:
        return f"错误: {e}"

model = ChatOpenAI(model="gpt-4.1")

agent = create_agent(
    model=model,
    tools=[get_weather, calculate],
    system_prompt="你是一个智能助手，可以查询天气和进行计算。"
)

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

print(result["messages"][-1].content)

带记忆的 Agent（多轮对话）

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver

model = ChatOpenAI(model="gpt-4.1")

agent = create_agent(
    model=model,
    tools=[get_weather, calculate],
    checkpointer=InMemorySaver(),  # 启用记忆
    system_prompt="你是一个有帮助的助手。"
)

# 配置线程 ID（区分不同对话）
config = {"configurable": {"thread_id": "conversation-1"}}

# 第一轮
result1 = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫小明"}]},
    config=config
)

# 第二轮（agent 会记住之前的内容）
result2 = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么名字？"}]},
    config=config
)

print(result2["messages"][-1].content)  # 应该回答 "小明"

模型配置

使用 init_chat_model（推荐）

from langchain.chat_models import init_chat_model

# 字符串方式
model = init_chat_model("gpt-4.1")
model = init_chat_model("gpt-4.1-mini", temperature=0.7)
model = init_chat_model("claude-sonnet-4-6")

# 带提供商前缀
model = init_chat_model("openai:gpt-4.1")
model = init_chat_model("anthropic:claude-sonnet-4-6")
model = init_chat_model("google:gemini-2.0-flash")

# 自定义配置
model = init_chat_model(
    "gpt-4.1",
    temperature=0.5,
    max_tokens=1000,
)

直接使用提供商类

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

# OpenAI
openai_model = ChatOpenAI(
    model="gpt-4.1",
    temperature=0.7,
)

# Anthropic
claude_model = ChatAnthropic(
    model="claude-sonnet-4-6",
    temperature=0.7,
)

流式输出

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

agent = create_agent(
    model=ChatOpenAI(model="gpt-4.1", streaming=True),
    tools=[get_weather, calculate],
)

config = {"configurable": {"thread_id": "stream-test"}}

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "介绍一下 Python 语言"}]},
    config=config
):
    if "messages" in chunk:
        for msg in chunk["messages"]:
            if hasattr(msg, "content"):
                print(msg.content, end="", flush=True)

错误处理

工具错误处理

@tool
def risky_operation(data: str) -> str:
    """可能失败的操作"""
    if "error" in data:
        raise ValueError("操作失败")
    return "成功"

Agent 错误处理

try:
    result = agent.invoke({
        "messages": [{"role": "user", "content": "..."}]
    })
except Exception as e:
    print(f"Agent 执行失败: {e}")

从旧 API 迁移

从 initialize_agent 迁移

旧代码：

from langchain.agents import initialize_agent, AgentType
from langchain.llms import OpenAI
from langchain.memory import ConversationBufferMemory

llm = OpenAI(temperature=0)
memory = ConversationBufferMemory(memory_key="chat_history")

agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,
    memory=memory,
)
result = agent.run("你好")

新代码：

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver

model = ChatOpenAI(model="gpt-4.1")

agent = create_agent(
    model=model,
    tools=tools,
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "chat-1"}}
result = agent.invoke(
    {"messages": [{"role": "user", "content": "你好"}]},
    config=config
)

从 AgentExecutor 迁移

旧代码：

from langchain.agents import AgentExecutor, create_tool_calling_agent

agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, memory=memory)
result = agent_executor.invoke({"input": "你好"})

新代码：

from langchain.agents import create_agent

agent = create_agent(
    model=model,
    tools=tools,
    checkpointer=InMemorySaver(),
)
result = agent.invoke({"messages": [{"role": "user", "content": "你好"}]})

小结

本章学习了：

• ✅ Agent 系统概念
• ✅ 新旧 API 对比
• ✅ 使用 @tool 定义工具
• ✅ 使用 create_agent 创建 Agent
• ✅ 带记忆的多轮对话
• ✅ 流式输出
• ✅ 从旧 API 迁移

下一章

第八章：RAG 应用 - 学习检索增强生成技术。

---

进阶内容：关于 Middleware 中间件、状态管理、长期记忆等高级特性，请参阅 第九章：Agent 进阶。