LangChain Agents 代理

Agents 是 LangChain 的核心功能，让 LLM 能够使用工具、做出决策，并完成复杂任务。

#什么是 Agent？

Agent 是一个能够：

• 理解用户意图
• 选择合适的工具
• 执行多步骤任务
• 根据结果调整策略

用户: "北京今天天气怎么样？适合户外运动吗？"
     ↓
Agent 思考: 需要先获取天气信息
     ↓
Agent 调用: get_weather("北京")
     ↓
Agent 收到: "晴，25°C，空气质量良好"
     ↓
Agent 回答: "北京今天晴朗，25°C，空气质量良好，非常适合户外运动！"

#快速开始

#创建简单 Agent

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

# 1. 定义工具
@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息"""
    # 模拟天气 API
    weather_data = {
        "北京": "晴，25°C",
        "上海": "多云，28°C"
    }
    return weather_data.get(city, f"{city}天气数据暂不可用")

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

# 2. 创建 Prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手，可以使用工具来回答问题。"),
    ("placeholder", "{chat_history}"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

# 3. 创建 Agent
llm = ChatOpenAI(model="gpt-4o")
tools = [get_weather, calculate]

agent = create_tool_calling_agent(llm, tools, prompt)

# 4. 创建 AgentExecutor
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 5. 运行
result = agent_executor.invoke({
    "input": "北京天气怎么样？另外帮我算一下 123 * 456"
})
print(result["output"])

#定义工具

#使用 @tool 装饰器

python复制
from langchain_core.tools import tool

@tool
def search_web(query: str) -> str:
    """搜索网络获取信息

    Args:
        query: 搜索关键词

    Returns:
        搜索结果摘要
    """
    # 实现搜索逻辑
    return f"关于 '{query}' 的搜索结果..."

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送电子邮件

    Args:
        to: 收件人邮箱
        subject: 邮件主题
        body: 邮件正文

    Returns:
        发送状态
    """
    # 实现发送逻辑
    return f"邮件已发送给 {to}"

#使用 StructuredTool

python复制
from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(description="搜索关键词")
    max_results: int = Field(default=5, description="最大结果数")

def search_function(query: str, max_results: int = 5) -> str:
    return f"搜索 '{query}'，返回 {max_results} 条结果"

search_tool = StructuredTool.from_function(
    func=search_function,
    name="search",
    description="搜索信息",
    args_schema=SearchInput
)

#内置工具

python复制
# 搜索工具
from langchain_community.tools import DuckDuckGoSearchRun
search = DuckDuckGoSearchRun()

# Wikipedia
from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())

# 计算器
from langchain_community.tools import LLMMathChain
math_tool = LLMMathChain.from_llm(llm)

# Python REPL
from langchain_experimental.tools import PythonREPLTool
python_repl = PythonREPLTool()

#Agent 类型

#Tool Calling Agent (推荐)

python复制
from langchain.agents import create_tool_calling_agent

agent = create_tool_calling_agent(llm, tools, prompt)

#ReAct Agent

python复制
from langchain.agents import create_react_agent
from langchain import hub

# 使用预定义的 ReAct prompt
prompt = hub.pull("hwchase17/react")

agent = create_react_agent(llm, tools, prompt)

#OpenAI Functions Agent

python复制
from langchain.agents import create_openai_functions_agent

agent = create_openai_functions_agent(llm, tools, prompt)

#Agent 配置

#AgentExecutor 选项

python复制
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,           # 显示详细日志
    max_iterations=10,      # 最大迭代次数
    max_execution_time=60,  # 最大执行时间（秒）
    early_stopping_method="generate",  # 早停方式
    handle_parsing_errors=True,  # 处理解析错误
    return_intermediate_steps=True  # 返回中间步骤
)

#流式输出

python复制
async for event in agent_executor.astream_events(
    {"input": "你的问题"},
    version="v1"
):
    kind = event["event"]
    if kind == "on_chat_model_stream":
        content = event["data"]["chunk"].content
        if content:
            print(content, end="")

#带记忆的 Agent

python复制
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

# 创建记忆存储
store = {}

def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

# 添加记忆
agent_with_history = RunnableWithMessageHistory(
    agent_executor,
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history"
)

# 使用（同一 session_id 会保持对话历史）
result = agent_with_history.invoke(
    {"input": "我叫小明"},
    config={"configurable": {"session_id": "user-123"}}
)

result = agent_with_history.invoke(
    {"input": "我叫什么名字？"},
    config={"configurable": {"session_id": "user-123"}}
)  # 会记住你叫小明

#实用示例

#研究助手 Agent

python复制
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from langchain_community.tools import DuckDuckGoSearchRun

# 工具
search = DuckDuckGoSearchRun()

@tool
def take_notes(content: str) -> str:
    """保存研究笔记"""
    with open("research_notes.txt", "a") as f:
        f.write(content + "\n\n")
    return "笔记已保存"

@tool
def read_notes() -> str:
    """读取之前的研究笔记"""
    try:
        with open("research_notes.txt", "r") as f:
            return f.read()
    except FileNotFoundError:
        return "暂无笔记"

# Prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个研究助手。你可以：
1. 搜索网络获取信息
2. 保存重要发现到笔记
3. 回顾之前的笔记

请系统性地进行研究，并记录重要发现。"""),
    ("placeholder", "{chat_history}"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

# 创建 Agent
llm = ChatOpenAI(model="gpt-4o")
tools = [search, take_notes, read_notes]

agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 使用
result = agent_executor.invoke({
    "input": "研究一下 2024 年 AI 发展趋势，并记录关键发现"
})

#代码助手 Agent

python复制
from langchain_experimental.tools import PythonREPLTool
from langchain_core.tools import tool

python_repl = PythonREPLTool()

@tool
def read_file(filepath: str) -> str:
    """读取文件内容"""
    with open(filepath, "r") as f:
        return f.read()

@tool
def write_file(filepath: str, content: str) -> str:
    """写入文件"""
    with open(filepath, "w") as f:
        f.write(content)
    return f"已写入 {filepath}"

prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个 Python 编程助手。你可以：
1. 执行 Python 代码
2. 读取和写入文件
3. 帮助用户完成编程任务

执行代码前请仔细检查，确保安全。"""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

tools = [python_repl, read_file, write_file]
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 使用
result = agent_executor.invoke({
    "input": "帮我写一个读取 CSV 文件并计算平均值的 Python 脚本"
})

#最佳实践

#1. 清晰的工具描述

python复制
@tool
def search(query: str) -> str:
    """搜索网络获取最新信息。

    适用场景：
    - 查找最新新闻和事件
    - 获取实时数据
    - 验证事实信息

    不适用于：
    - 需要深度分析的问题
    - 历史文档查询

    Args:
        query: 搜索关键词，尽量具体明确

    Returns:
        搜索结果的摘要
    """
    pass

#2. 限制迭代次数

python复制
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=5,  # 防止无限循环
    max_execution_time=30  # 超时限制
)

#3. 错误处理

python复制
@tool
def risky_operation(param: str) -> str:
    """可能出错的操作"""
    try:
        # 执行操作
        return "成功"
    except Exception as e:
        return f"操作失败: {str(e)}"

#4. 安全考虑

python复制
# 限制代码执行
python_repl = PythonREPLTool(
    # 在沙箱环境中运行
)

# 验证输入
@tool
def safe_search(query: str) -> str:
    """安全搜索"""
    # 过滤敏感内容
    if any(bad in query.lower() for bad in ["hack", "attack"]):
        return "搜索被拒绝"
    return do_search(query)

#下一步

• RAG 检索增强 - 知识库问答
• LangGraph - 复杂工作流
• Tool Use - Claude 工具调用

---

提示：Agent 功能强大但需要谨慎使用，尤其是涉及代码执行和文件操作时。