跳到主要内容

LangChain Agent 实战

LangChain 是目前最流行的 AI Agent 开发框架之一，它提供了丰富的组件和工具，帮助开发者快速构建智能代理应用。本章将通过实际案例，带你掌握 LangChain Agent 的开发技巧。

环境准备​

首先安装必要的依赖：

pip install langchain langchain-openai langchain-community

配置 API Key：

import os

os.environ["OPENAI_API_KEY"] = "your-api-key"

第一个 Agent​

让我们从最简单的 Agent 开始，一个能进行数学计算的助手：

from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain import hub
from langchain_core.tools import tool

@tool
def add(a: int, b: int) -> int:
    """将两个数字相加"""
    return a + b

@tool
def subtract(a: int, b: int) -> int:
    """将两个数字相减"""
    return a - b

@tool
def multiply(a: int, b: int) -> int:
    """将两个数字相乘"""
    return a * b

@tool
def divide(a: float, b: float) -> float:
    """将两个数字相除"""
    if b == 0:
        return "错误：除数不能为零"
    return a / b

tools = [add, subtract, multiply, divide]

llm = ChatOpenAI(model="gpt-4", temperature=0)

prompt = hub.pull("hwchase17/react")

agent = create_react_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    handle_parsing_errors=True
)

result = agent_executor.invoke({
    "input": "计算 (23 + 17) * 5 - 10 的结果"
})

print(f"最终答案：{result['output']}")

运行这个 Agent，你会看到它如何一步步思考并调用工具：

> Entering new AgentExecutor chain...
思考：我需要先计算 23 + 17，然后用结果乘以 5，最后减去 10
行动：add
行动输入：{"a": 23, "b": 17}
观察：40
思考：现在我需要将 40 乘以 5
行动：multiply
行动输入：{"a": 40, "b": 5}
观察：200
思考：最后减去 10
行动：subtract
行动输入：{"a": 200, "b": 10}
观察：190
思考：我已经得到了最终答案
最终答案：结果是 190

使用 Function Calling Agent​

Function Calling Agent 比 ReAct Agent 更高效，推荐使用：

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

@tool
def get_current_weather(city: str) -> str:
    """
    获取指定城市的当前天气
    
    参数：
    city: 城市名称，如"北京"、"上海"
    """
    weather_data = {
        "北京": "晴，温度 22°C，空气质量良好",
        "上海": "多云，温度 26°C，有轻微雾霾",
        "广州": "小雨，温度 28°C，湿度较高",
        "深圳": "晴转多云，温度 30°C",
    }
    return weather_data.get(city, f"未找到 {city} 的天气信息")

@tool
def get_weather_forecast(city: str, days: int = 3) -> str:
    """
    获取未来几天的天气预报
    
    参数：
    city: 城市名称
    days: 预报天数，默认3天
    """
    return f"{city} 未来 {days} 天天气预报：第一天晴，第二天多云，第三天小雨"

tools = [get_current_weather, get_weather_forecast]

llm = ChatOpenAI(model="gpt-4", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个天气助手，可以帮助用户查询天气信息。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True
)

result = agent_executor.invoke({
    "input": "北京今天天气怎么样？明天呢？"
})

print(result["output"])

构建多功能助手​

让我们构建一个更复杂的 Agent，集成多种工具：

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

@tool
def web_search(query: str) -> str:
    """
    在互联网上搜索信息
    
    参数：
    query: 搜索关键词
    """
    return f"搜索 '{query}' 的结果：找到相关信息..."

@tool
def get_weather(city: str) -> str:
    """
    获取城市天气信息
    
    参数：
    city: 城市名称
    """
    weather_data = {
        "北京": {"temp": 22, "condition": "晴"},
        "上海": {"temp": 26, "condition": "多云"},
    }
    data = weather_data.get(city, {"temp": 20, "condition": "未知"})
    return f"{city}：{data['condition']}，温度 {data['temp']}°C"

@tool
def calculate(expression: str) -> str:
    """
    计算数学表达式
    
    参数：
    expression: 数学表达式，如 "2 + 3 * 4"
    """
    try:
        allowed_chars = set("0123456789+-*/.() ")
        if not all(c in allowed_chars for c in expression):
            return "错误：表达式包含非法字符"
        result = eval(expression)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

@tool
def get_current_time(timezone: str = "Asia/Shanghai") -> str:
    """
    获取当前时间
    
    参数：
    timezone: 时区，默认为上海时区
    """
    now = datetime.now()
    return f"当前时间：{now.strftime('%Y-%m-%d %H:%M:%S')}"

@tool
def translate(text: str, target_lang: str = "英语") -> str:
    """
    翻译文本
    
    参数：
    text: 要翻译的文本
    target_lang: 目标语言，如"英语"、"日语"
    """
    return f"[翻译结果-{target_lang}]：{text}"

tools = [
    web_search,
    get_weather,
    calculate,
    get_current_time,
    translate
]

llm = ChatOpenAI(model="gpt-4", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个智能助手，可以帮助用户完成多种任务：
    
    1. 搜索互联网信息
    2. 查询天气
    3. 数学计算
    4. 查看时间
    5. 翻译文本
    
    请根据用户的需求选择合适的工具完成任务。
    如果用户的问题不需要使用工具，直接回答即可。"""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10
)

def chat(user_input: str) -> str:
    result = agent_executor.invoke({"input": user_input})
    return result["output"]

print(chat("北京今天天气怎么样？"))
print(chat("帮我计算 123 * 456"))
print(chat("现在几点了？"))
print(chat("把'你好，世界'翻译成英语"))

添加记忆功能​

让 Agent 能够记住对话历史：

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

@tool
def save_note(content: str) -> str:
    """保存笔记"""
    return f"已保存笔记：{content[:50]}..."

@tool
def search_notes(query: str) -> str:
    """搜索笔记"""
    return f"找到与 '{query}' 相关的笔记..."

tools = [save_note, search_notes]

llm = ChatOpenAI(model="gpt-4", temperature=0)

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个笔记助手，可以帮助用户管理笔记。"),
    ("placeholder", "{chat_history}"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True
)

agent_executor.invoke({"input": "帮我保存一条笔记：明天下午3点开会"})
agent_executor.invoke({"input": "我刚才保存了什么笔记？"})

使用 LangGraph 构建复杂 Agent​

LangGraph 是 LangChain 的新一代 Agent 框架，支持更复杂的控制流：

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

@tool
def search(query: str) -> str:
    """搜索信息"""
    return f"搜索结果：{query}"

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

tools = [search, calculator]
tool_executor = ToolExecutor(tools)

llm = ChatOpenAI(model="gpt-4", temperature=0)
llm_with_tools = llm.bind_tools(tools)

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]

def should_continue(state: AgentState):
    messages = state["messages"]
    last_message = messages[-1]
    
    if not last_message.tool_calls:
        return "end"
    return "continue"

def call_model(state: AgentState):
    messages = state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

def call_tool(state: AgentState):
    messages = state["messages"]
    last_message = messages[-1]
    
    tool_calls = last_message.tool_calls
    
    tool_messages = []
    for tool_call in tool_calls:
        tool_message = tool_executor.invoke(
            ToolInvocation(
                tool=tool_call["name"],
                tool_input=tool_call["args"]
            )
        )
        tool_messages.append(tool_message)
    
    return {"messages": tool_messages}

workflow = StateGraph(AgentState)

workflow.add_node("agent", call_model)
workflow.add_node("action", call_tool)

workflow.set_entry_point("agent")

workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "action",
        "end": END
    }
)

workflow.add_edge("action", "agent")

app = workflow.compile()

from langchain_core.messages import HumanMessage

inputs = {"messages": [HumanMessage(content="搜索 Python 教程，然后计算 2 的 10 次方")]}
result = app.invoke(inputs)

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

结构化输出​

让 Agent 返回结构化的结果：

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from typing import List

class SearchResult(BaseModel):
    """搜索结果"""
    query: str = Field(description="搜索查询")
    results: List[str] = Field(description="搜索结果列表")
    total: int = Field(description="结果总数")

class AnalysisResult(BaseModel):
    """分析结果"""
    summary: str = Field(description="摘要")
    key_points: List[str] = Field(description="关键点")
    sentiment: str = Field(description="情感倾向")

@tool
def analyze_text(text: str) -> str:
    """分析文本内容"""
    return f"分析结果：{text[:100]}..."

tools = [analyze_text]

llm = ChatOpenAI(model="gpt-4", temperature=0)

structured_llm = llm.with_structured_output(AnalysisResult)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个文本分析助手。"),
    ("human", "{input}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True
)

result = agent_executor.invoke({
    "input": "分析这段文本：今天天气真好，阳光明媚，心情很愉快。"
})

print(result["output"])

错误处理与重试​

健壮的 Agent 需要处理各种错误情况：

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate
from tenacity import retry, stop_after_attempt, wait_exponential

@tool
def unreliable_api(query: str) -> str:
    """可能失败的 API 调用"""
    import random
    if random.random() < 0.5:
        raise Exception("API 调用失败")
    return f"API 结果：{query}"

tools = [unreliable_api]

llm = ChatOpenAI(model="gpt-4", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个助手。如果工具调用失败，请尝试其他方法或告知用户。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    handle_parsing_errors=True,
    max_iterations=5,
    early_stopping_method="generate"
)

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def run_agent_with_retry(input_text: str):
    return agent_executor.invoke({"input": input_text})

try:
    result = run_agent_with_retry("调用 API 查询信息")
    print(result["output"])
except Exception as e:
    print(f"Agent 执行失败：{e}")

完整示例：智能研究助手​

让我们构建一个完整的研究助手 Agent：

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain.memory import ConversationBufferMemory
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate
from typing import List, Dict
import json

notes_db: Dict[str, str] = {}

@tool
def search_web(query: str) -> str:
    """
    在互联网上搜索信息
    
    参数：
    query: 搜索关键词
    """
    mock_results = [
        {"title": f"{query} - 概述", "snippet": f"关于 {query} 的基本概念和定义..."},
        {"title": f"{query} - 深入研究", "snippet": f"{query} 的最新研究进展..."},
        {"title": f"{query} - 应用案例", "snippet": f"{query} 在实际中的应用..."},
    ]
    return json.dumps(mock_results, ensure_ascii=False)

@tool
def save_note(title: str, content: str) -> str:
    """
    保存研究笔记
    
    参数：
    title: 笔记标题
    content: 笔记内容
    """
    notes_db[title] = content
    return f"已保存笔记《{title}》"

@tool
def list_notes() -> str:
    """列出所有保存的笔记"""
    if not notes_db:
        return "暂无保存的笔记"
    return json.dumps(
        [{"title": title, "preview": content[:50] + "..."} for title, content in notes_db.items()],
        ensure_ascii=False
    )

@tool
def read_note(title: str) -> str:
    """
    读取指定笔记
    
    参数：
    title: 笔记标题
    """
    if title in notes_db:
        return notes_db[title]
    return f"未找到笔记《{title}》"

@tool
def summarize(text: str) -> str:
    """
    总结文本内容
    
    参数：
    text: 要总结的文本
    """
    return f"【摘要】{text[:100]}..."

tools = [search_web, save_note, list_notes, read_note, summarize]

llm = ChatOpenAI(model="gpt-4", temperature=0)

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个智能研究助手，可以帮助用户：
    
    1. 搜索和收集信息
    2. 整理和保存研究笔记
    3. 总结和分析内容
    
    请根据用户需求灵活使用工具，提供专业的研究支持。
    
    当前保存的笔记：{notes}"""),
    ("placeholder", "{chat_history}"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True,
    max_iterations=10
)

def research_assistant(user_input: str) -> str:
    notes_preview = ", ".join(notes_db.keys()) if notes_db else "暂无"
    result = agent_executor.invoke({
        "input": user_input,
        "notes": notes_preview
    })
    return result["output"]

print(research_assistant("帮我搜索关于人工智能的最新进展"))
print(research_assistant("把这些信息保存为笔记，标题是'AI进展'"))
print(research_assistant("我有哪些笔记？"))
print(research_assistant("读取'AI进展'这篇笔记并总结"))

小结​

本章介绍了使用 LangChain 构建 AI Agent 的核心技能：

  • 使用 ReAct Agent 和 Function Calling Agent 两种模式
  • 定义和注册自定义工具
  • 为 Agent 添加记忆功能
  • 使用 LangGraph 构建复杂控制流
  • 实现结构化输出和错误处理
  • 构建完整的实用 Agent 应用

下一章我们将学习多 Agent 协作的高级主题。

参考资料​