LangChain 1.0 Agent系统开发指南：从消息系统到工具调用

1. LangChain 1.0 Agent系统快速入门指南

LangChain 1.0带来了全新的Agent系统构建方式，让开发者能够快速搭建具备自主决策能力的AI助手。本文将带你从零开始，掌握LangChain 1.0中最核心的消息系统和工具调用机制。

1.1 环境准备与版本要求

在开始之前，请确保你的开发环境满足以下要求：

• Python 3.10或更高版本
• LangChain 1.0.7+
• LangGraph 1.0.3+

安装依赖包：

bash复制pip install langchain langchain-openai langchain-core python-dotenv

1.2 消息系统基础

LangChain 1.0引入了统一的消息类型系统，这是构建Agent对话的基础。让我们先了解四种核心消息类型：

1. HumanMessage：表示用户输入
2. AIMessage：表示AI模型的回复
3. SystemMessage：定义AI行为和角色
4. ToolMessage：表示工具执行的返回结果

1.2.1 消息类型详解

HumanMessage示例：

python复制from langchain_core.messages import HumanMessage

# 简单文本消息
message = HumanMessage(content="什么是LangChain？")

# 带元数据的消息
message = HumanMessage(
    content="分析这张图片",
    metadata={"user_id": "123"}
)

AIMessage示例：

python复制from langchain_core.messages import AIMessage

# 简单文本回复
response = AIMessage(content="LangChain是一个框架...")

# 带工具调用的回复
response = AIMessage(
    content="",
    tool_calls=[{
        "id": "call_123",
        "name": "search",
        "args": {"query": "LangChain"}
    }]
)

SystemMessage示例：

python复制from langchain_core.messages import SystemMessage

system = SystemMessage(
    content="""你是一个有帮助的AI助手。
    回答要简洁准确。
    必要时使用工具。"""
)

ToolMessage示例：

python复制from langchain_core.messages import ToolMessage

tool_result = ToolMessage(
    content="搜索结果：...",
    tool_call_id="call_123"
)

1.3 Content Blocks创新

Content Blocks是LangChain 1.0的重大创新，提供了跨Provider的统一内容表示方式。它解决了以下问题：

1. 不同模型的输出格式不一致
2. 无法统一处理工具调用、思考过程、多模态内容
3. Provider切换需要重写代码

1.3.1 Content Blocks类型

1. Text Block：基础文本内容
2. Tool Use Block：统一的工具调用格式
3. Thinking Block：思考过程（Claude模型特有）
4. Image Block：图像内容
5. Citation Block：引用来源（Gemini模型特有）

Text Block示例：

python复制response = AIMessage(
    content=[{"type": "text", "text": "这是回答..."}]
)

Tool Use Block示例：

python复制response = AIMessage(
    content=[{
        "type": "tool_use",
        "id": "call_123",
        "name": "search",
        "input": {"query": "LangChain"}
    }]
)

1.4 工具定义与调用

LangChain提供了三种定义工具的方式：

1. @tool装饰器：最简单快速的方式
2. StructuredTool类：更灵活的定义方式
3. BaseTool继承：完全自定义控制

1.4.1 @tool装饰器示例

python复制from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """搜索网络获取信息。
    
    Args:
        query: 搜索查询字符串
    """
    return f"搜索结果：{query}"

1.4.2 带复杂参数的工具

python复制from typing import Optional
from pydantic import Field

@tool
def advanced_search(
    query: str = Field(description="搜索查询"),
    max_results: int = Field(default=10, description="最大结果数"),
    language: Optional[str] = Field(default="en", description="语言")
) -> str:
    """带过滤条件的高级搜索"""
    return f"找到{max_results}条'{query}'的结果，语言：{language}"

1.5 创建第一个Agent

现在，让我们创建一个简单的Agent，它能够搜索维基百科和执行数学计算。

1.5.1 定义工具

python复制from langchain_core.tools import tool
import requests
import math

@tool
def search_wikipedia(query: str) -> str:
    """搜索维基百科获取信息"""
    url = "https://en.wikipedia.org/w/api.php"
    params = {
        "action": "query",
        "list": "search",
        "srsearch": query,
        "format": "json"
    }
    response = requests.get(url, params=params)
    data = response.json()
    return data["query"]["search"][0]["snippet"] if data["query"]["search"] else "无结果"

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression, {"__builtins__": None}, {
            "sqrt": math.sqrt,
            "sin": math.sin,
            "cos": math.cos,
            "pi": math.pi
        })
        return str(result)
    except Exception as e:
        return f"错误：{str(e)}"

1.5.2 创建并运行Agent

python复制from langchain_openai import ChatOpenAI
from langchain.agents import create_agent

# 1. 配置LLM模型
model = ChatOpenAI(model="gpt-4", temperature=0)

# 2. 定义工具列表
tools = [search_wikipedia, calculate]

# 3. 创建Agent
agent = create_agent(model, tools)

# 4. 运行Agent
result = agent.invoke({
    "messages": [("user", "计算圆周率平方根与2的乘积")]
})

# 5. 获取最终结果
print(result["messages"][-1].content)

1.6 错误处理策略

在实际应用中，完善的错误处理机制至关重要。以下是几种常见的错误处理策略：

1. 基础错误处理：捕获并返回错误信息
2. 结构化错误返回：使用Pydantic模型定义错误格式
3. 重试机制：使用tenacity库实现自动重试
4. 超时控制：使用asyncio.wait_for设置超时

结构化错误示例：

python复制from typing import Union, Optional
from pydantic import BaseModel, Field

class ToolResult(BaseModel):
    """工具执行结果的结构化表示"""
    success: bool = Field(description="执行是否成功")
    data: Optional[Union[str, dict]] = Field(default=None, description="成功时的数据")
    error: Optional[str] = Field(default=None, description="失败时的错误信息")

@tool
def safe_search(query: str) -> str:
    """带错误处理的搜索"""
    try:
        results = f"搜索'{query}'的结果"
        return ToolResult(success=True, data=results).model_dump_json()
    except Exception as e:
        return ToolResult(success=False, error=str(e)).model_dump_json()

1.7 生产环境最佳实践

在实际生产环境中部署Agent时，需要考虑以下因素：

1. 性能优化：合理设置recursion_limit防止无限循环
2. 日志记录：使用Callback Handler监控执行过程
3. 错误恢复：实现完善的错误处理策略
4. 安全防护：对用户输入进行验证和过滤

性能优化示例：

python复制from langgraph.errors import GraphRecursionError

try:
    result = agent.invoke(
        {"messages": [("user", "复杂问题")]},
        config={"recursion_limit": 11}  # 最多5次迭代
    )
except GraphRecursionError as e:
    print(f"达到最大迭代次数：{e}")

1.8 总结与进阶学习

通过本文，你已经掌握了LangChain 1.0中构建Agent系统的核心概念和基本方法。以下是关键要点总结：

1. 理解四种核心消息类型及其应用场景
2. 掌握Content Blocks的创新设计
3. 学会三种定义工具的方式
4. 能够创建并运行简单的Agent
5. 了解基本的错误处理策略

对于想要深入学习的开发者，建议进一步探索以下主题：

1. Middleware机制：在Agent执行过程中注入自定义逻辑
2. 结构化输出：使用response_format参数控制输出格式
3. 多模态处理：处理图像、音频等非文本内容
4. 高级工具调用：实现并行工具调用和复杂参数验证

在实际项目中，建议从简单场景开始，逐步增加复杂度，并始终关注性能和安全性问题。LangChain提供了丰富的文档和社区支持，遇到问题时可以查阅官方文档或参与社区讨论。