---

关于作者

• 深耕领域：大语言模型开发 / RAG 知识库 / AI Agent 落地 / 模型微调
• 技术栈：Python | RAG (LangChain / Dify + Milvus) | FastAPI + Docker
• 工程能力：专注模型工程化部署、知识库构建与优化，擅长全流程解决方案

「让 AI 交互更智能，让技术落地更高效」
欢迎技术探讨与项目合作，解锁大模型与智能交互的无限可能！

---

LangChain Agents 与 Tools 代理工具深度解析

摘要

Agent（代理）是 LangChain 最强大的特性之一，它赋予 LLM 自主决策和工具使用的能力。本文将深入解析 LangChain 的 Agent 架构、Tool 定义与实现、ReAct 推理模式、以及如何构建生产级的智能代理系统。

---

一、Agent 概述

1.1 什么是 Agent

Agent 是一个能够自主决策、选择工具、执行任务的智能体。与传统的 Chain 不同，Agent 可以根据输入动态决定执行路径。

1.2 Agent 核心组件

1.3 Agent 类型

类型	说明	适用场景
ReAct Agent	推理-行动循环	通用任务、工具调用
OpenAI Tools Agent	OpenAI 原生工具调用	GPT 模型、高效执行
Structured Tool Chat	结构化多参数工具	复杂工具、多参数
Self-Ask	自我提问分解	复杂问题分解
Plan-and-Execute	规划后执行	长时任务、多步骤

---

二、Tool 工具系统

2.1 Tool 基础定义

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息。
  
    Args:
        city: 城市名称，如"北京"、"上海"
  
    Returns:
        天气信息字符串
    """
    # 实际实现会调用天气 API
    weather_data = {
        "北京": "晴天，温度 25°C，空气质量良好",
        "上海": "多云，温度 28°C，有轻微雾霾",
        "深圳": "小雨，温度 30°C，湿度较高",
    }
    return weather_data.get(city, f"未找到 {city} 的天气信息")

@tool
def calculate(expression: str) -> str:
    """计算数学表达式。
  
    Args:
        expression: 数学表达式，如"2+3*4"
  
    Returns:
        计算结果
    """
    try:
        result = eval(expression)
        return f"计算结果: {result}"
    except Exception as e:
        return f"计算错误: {e}"

# 工具列表
tools = [get_weather, calculate]

2.2 Tool 装饰器详解

from langchain_core.tools import tool
from typing import Optional

@tool
def search_web(
    query: str,
    num_results: int = 5,
    include_images: bool = False,
) -> str:
    """搜索网络获取信息。
  
    这是一个强大的网络搜索工具，可以搜索互联网上的各种信息。
  
    Args:
        query: 搜索关键词
        num_results: 返回结果数量，默认 5 条
        include_images: 是否包含图片结果
  
    Returns:
        搜索结果字符串
    """
    # 实际实现
    return f"搜索 '{query}' 找到 {num_results} 条结果"

# 查看工具定义
print(search_web.name)          # search_web
print(search_web.description)   # 自动从 docstring 提取
print(search_web.args_schema)   # 自动生成的参数 Schema

2.3 StructuredTool 类

from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    """搜索输入参数。"""
    query: str = Field(description="搜索关键词")
    num_results: int = Field(default=5, description="结果数量")
    region: str = Field(default="cn", description="搜索区域")

def search_function(query: str, num_results: int = 5, region: str = "cn") -> str:
    """实际搜索函数。"""
    return f"在 {region} 搜索 '{query}'，返回 {num_results} 条结果"

# 创建结构化工具
search_tool = StructuredTool.from_function(
    func=search_function,
    name="web_search",
    description="搜索网络获取信息",
    args_schema=SearchInput,
)

# 使用
result = search_tool.invoke({
    "query": "Python 教程",
    "num_results": 10,
    "region": "us"
})

2.4 异步工具

from langchain_core.tools import tool
import aiohttp

@tool
async def async_http_get(url: str) -> str:
    """异步 HTTP GET 请求。
  
    Args:
        url: 请求的 URL 地址
  
    Returns:
        响应内容
    """
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

# 异步调用
import asyncio
result = asyncio.run(async_http_get.ainvoke({"url": "https://api.github.com"}))

2.5 工具错误处理

from langchain_core.tools import ToolException, tool

@tool
def risky_operation(data: str) -> str:
    """可能失败的操作。
  
    Args:
        data: 输入数据
  
    Returns:
        处理结果
  
    Raises:
        ToolException: 当数据无效时抛出
    """
    if not data:
        raise ToolException("输入数据不能为空")
  
    if len(data) > 1000:
        raise ToolException("数据长度超过限制（最大 1000 字符）")
  
    return f"处理成功: {data}"

# 配置错误处理
from langchain_core.tools import Tool

tool_with_error_handling = Tool(
    name="safe_operation",
    func=risky_operation,
    description="安全的操作",
    handle_tool_error=True,  # 自动处理错误，返回错误信息而非抛出异常
)

---

三、ReAct Agent

3.1 ReAct 模式

ReAct（Reasoning + Acting）是一种让 LLM 交替进行推理和行动的模式：

3.2 构建 ReAct Agent

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

# 定义工具
@tool
def search(query: str) -> str:
    """搜索网络获取信息。"""
    return f"搜索结果: {query} 相关信息..."

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

tools = [search, calculator]

# 定义 Prompt
prompt = PromptTemplate.from_template("""你是一个有用的助手，可以使用工具来回答问题。

你可以使用以下工具:
{tools}

使用以下格式回答：

Question: 用户的问题
Thought: 你应该思考做什么
Action: 要使用的工具，必须是 [{tool_names}] 之一
Action Input: 工具的输入
Observation: 工具的输出
... (这个 Thought/Action/Action Input/Observation 可以重复 N 次)
Thought: 我现在知道最终答案了
Final Answer: 最终答案

开始！

Question: {input}
Thought: {agent_scratchpad}""")

# 创建 Agent
llm = ChatOpenAI(model="gpt-4", temperature=0)
agent = create_react_agent(llm, tools, prompt)

# 创建 AgentExecutor
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,  # 打印推理过程
    handle_parsing_errors=True,  # 处理解析错误
    max_iterations=5,  # 最大迭代次数
)

# 运行
result = agent_executor.invoke({
    "input": "搜索 Python 最新版本，然后计算 3.12 + 0.1"
})
print(result["output"])

3.3 AgentExecutor 参数

参数	类型	说明
agent	Agent	代理实例
tools	list	工具列表
verbose	bool	是否打印推理过程
max_iterations	int	最大迭代次数
max_execution_time	float	最大执行时间（秒）
handle_parsing_errors	bool	是否处理解析错误
return_intermediate_steps	bool	是否返回中间步骤
early_stopping_method	str	提前停止方式

---

四、OpenAI Tools Agent

4.1 OpenAI 原生工具调用

OpenAI 模型原生支持工具调用，比 ReAct 更高效：

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

# 定义工具
@tool
def get_current_weather(location: str, unit: str = "celsius") -> str:
    """获取指定位置的当前天气。
  
    Args:
        location: 城市名称
        unit: 温度单位，celsius 或 fahrenheit
    """
    return f"{location} 当前温度 25°{unit[0].upper()}"

@tool
def convert_currency(amount: float, from_currency: str, to_currency: str) -> str:
    """货币转换。
  
    Args:
        amount: 金额
        from_currency: 源货币代码
        to_currency: 目标货币代码
    """
    rates = {"USD": 7.2, "EUR": 7.8, "JPY": 0.048}
    if from_currency == "CNY":
        result = amount / rates.get(to_currency, 1)
    else:
        result = amount * rates.get(from_currency, 1)
    return f"{amount} {from_currency} = {result:.2f} {to_currency}"

tools = [get_current_weather, convert_currency]

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

# 创建 Agent
llm = ChatOpenAI(model="gpt-4o")
agent = create_tool_calling_agent(llm, tools, prompt)

# 创建执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
)

# 运行
result = agent_executor.invoke({
    "input": "北京现在天气怎么样？另外 100 美元等于多少人民币？"
})
print(result["output"])

4.2 流式输出

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

@tool
def search(query: str) -> str:
    """搜索网络。"""
    return f"搜索结果: {query}"

tools = [search]
llm = ChatOpenAI(model="gpt-4o")
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, stream_runnable=True)

# 流式输出
for chunk in agent_executor.stream({"input": "搜索 Python 教程"}):
    if "actions" in chunk:
        for action in chunk["actions"]:
            print(f"[工具调用] {action.tool}: {action.tool_input}")
    elif "steps" in chunk:
        for step in chunk["steps"]:
            print(f"[工具结果] {step.observation}")
    elif "output" in chunk:
        print(f"[最终答案] {chunk['output']}")

---

五、自定义 Agent

5.1 继承 BaseSingleActionAgent

from typing import Any, List, Tuple, Union
from langchain.agents import BaseSingleActionAgent
from langchain.schema import AgentAction, AgentFinish
from langchain_core.tools import BaseTool
from langchain_openai import ChatOpenAI
from pydantic import Field

class CustomAgent(BaseSingleActionAgent):
    """自定义单动作 Agent。"""
  
    llm: ChatOpenAI = Field(...)
    tools: List[BaseTool] = Field(default_factory=list)
  
    @property
    def input_keys(self) -> List[str]:
        return ["input"]
  
    def plan(
        self,
        intermediate_steps: List[Tuple[AgentAction, str]],
        **kwargs: Any,
    ) -> Union[AgentAction, AgentFinish]:
        """规划下一步行动。"""
        user_input = kwargs["input"]
      
        # 如果没有中间步骤，选择工具
        if not intermediate_steps:
            # 使用 LLM 决定使用哪个工具
            tool_names = [t.name for t in self.tools]
            prompt = f"""用户输入: {user_input}
可用工具: {tool_names}
请决定使用哪个工具，返回工具名称。如果不需要工具，返回 'FINISH'。"""
          
            response = self.llm.invoke(prompt)
            tool_name = response.content.strip()
          
            if tool_name == "FINISH":
                return AgentFinish(
                    return_values={"output": "我无法回答这个问题"},
                    log="无需使用工具",
                )
          
            return AgentAction(
                tool=tool_name,
                tool_input=user_input,
                log=f"选择工具: {tool_name}",
            )
      
        # 有中间步骤，生成最终答案
        last_action, last_observation = intermediate_steps[-1]
        return AgentFinish(
            return_values={"output": last_observation},
            log="使用工具完成",
        )
  
    async def aplan(
        self,
        intermediate_steps: List[Tuple[AgentAction, str]],
        **kwargs: Any,
    ) -> Union[AgentAction, AgentFinish]:
        """异步规划。"""
        return self.plan(intermediate_steps, **kwargs)

5.2 使用 LCEL 构建 Agent

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 定义工具
@tool
def search(query: str) -> str:
    """搜索网络。"""
    return f"搜索: {query}"

@tool  
def calculate(expression: str) -> str:
    """计算。"""
    return str(eval(expression))

tools = [search, calculate]
tools_map = {t.name: t for t in tools}

# 绑定工具到模型
llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)

# 定义 Agent 循环
async def agent_loop(prompt: str, max_iterations: int = 5):
    """Agent 执行循环。"""
    messages = [{"role": "user", "content": prompt}]
  
    for _ in range(max_iterations):
        # 调用模型
        response = await llm.ainvoke(messages)
      
        # 无工具调用则返回
        if not response.tool_calls:
            return response.content
      
        # 添加 AI 响应
        messages.append(response)
      
        # 执行工具
        for tool_call in response.tool_calls:
            tool = tools_map[tool_call["name"]]
            result = await tool.ainvoke(tool_call["args"])
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call["id"],
                "content": result,
            })
  
    return "达到最大迭代次数"

# 使用
import asyncio
result = asyncio.run(agent_loop("搜索 Python 最新版本"))

---

六、工具箱与工具组合

6.1 Tool 组合

from langchain_core.tools import tool

# 基础工具
@tool
def read_file(path: str) -> str:
    """读取文件内容。"""
    with open(path, "r", encoding="utf-8") as f:
        return f.read()

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

@tool
def list_directory(path: str) -> str:
    """列出目录内容。"""
    import os
    return "\n".join(os.listdir(path))

# 组合成工具箱
file_tools = [read_file, write_file, list_directory]

# 可以按需组合
@tool
def search_web(query: str) -> str:
    """搜索网络。"""
    return f"搜索结果: {query}"

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件。"""
    return f"邮件已发送给 {to}"

# 完整工具集
all_tools = file_tools + [search_web, send_email]

6.2 工具分组

from langchain_core.tools import BaseToolkit, StructuredTool

class FileToolkit(BaseToolkit):
    """文件操作工具箱。"""
  
    def get_tools(self) -> list:
        """返回工具列表。"""
        return [
            StructuredTool.from_function(
                func=self._read_file,
                name="read_file",
                description="读取文件内容",
            ),
            StructuredTool.from_function(
                func=self._write_file,
                name="write_file",
                description="写入文件",
            ),
            StructuredTool.from_function(
                func=self._list_dir,
                name="list_directory",
                description="列出目录",
            ),
        ]
  
    def _read_file(self, path: str) -> str:
        with open(path, "r", encoding="utf-8") as f:
            return f.read()
  
    def _write_file(self, path: str, content: str) -> str:
        with open(path, "w", encoding="utf-8") as f:
            f.write(content)
        return f"已写入: {path}"
  
    def _list_dir(self, path: str) -> str:
        import os
        return "\n".join(os.listdir(path))

# 使用工具箱
toolkit = FileToolkit()
tools = toolkit.get_tools()

6.3 动态工具加载

from langchain_core.tools import StructuredTool
from typing import Callable

class ToolRegistry:
    """工具注册中心。"""
  
    def __init__(self):
        self._tools: dict[str, StructuredTool] = {}
  
    def register(self, name: str, func: Callable, description: str):
        """注册工具。"""
        self._tools[name] = StructuredTool.from_function(
            func=func,
            name=name,
            description=description,
        )
  
    def get_tool(self, name: str) -> StructuredTool | None:
        """获取工具。"""
        return self._tools.get(name)
  
    def get_all_tools(self) -> list[StructuredTool]:
        """获取所有工具。"""
        return list(self._tools.values())
  
    def unregister(self, name: str):
        """注销工具。"""
        self._tools.pop(name, None)

# 使用
registry = ToolRegistry()

registry.register(
    name="weather",
    func=lambda city: f"{city} 天气晴",
    description="获取天气",
)

registry.register(
    name="calculator",
    func=lambda expr: str(eval(expr)),
    description="计算表达式",
)

tools = registry.get_all_tools()

---

七、高级特性

7.1 工具返回 Artifact

from langchain_core.tools import tool, ToolException
from typing import Any

@tool(response_format="content_and_artifact")
def generate_image(prompt: str) -> tuple[str, Any]:
    """生成图片。
  
    Args:
        prompt: 图片描述
  
    Returns:
        元组：(文本描述, 图片数据)
    """
    # 模拟图片生成
    image_data = b"fake_image_data"
  
    return (
        f"已生成图片: {prompt}",
        image_data,  # artifact，不会发送给 LLM
    )

# 调用
result = generate_image.invoke({"prompt": "一只猫"})
print(result.content)     # 文本描述
print(result.artifact)    # 图片数据（二进制）

7.2 工具调用钩子

from langchain_core.tools import tool
from langchain_core.callbacks import BaseCallbackHandler

class ToolCallbackHandler(BaseCallbackHandler):
    """工具调用回调。"""
  
    def on_tool_start(self, serialized, input_str, **kwargs):
        print(f"[工具开始] {serialized.get('name')}: {input_str}")
  
    def on_tool_end(self, output, **kwargs):
        print(f"[工具结束] 输出: {output}")
  
    def on_tool_error(self, error, **kwargs):
        print(f"[工具错误] {error}")

@tool
def my_tool(query: str) -> str:
    """示例工具。"""
    return f"处理: {query}"

# 使用回调
result = my_tool.invoke(
    {"query": "test"},
    config={"callbacks": [ToolCallbackHandler()]},
)

7.3 人机协作

from langchain_core.tools import tool
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

@tool
def human_input(prompt: str) -> str:
    """请求人类输入。
  
    当 Agent 需要人类确认或额外信息时使用。
  
    Args:
        prompt: 向人类提问的内容
  
    Returns:
        人类的回答
    """
    return input(f"[Agent 提问] {prompt}: ")

# 构建 Agent
from langchain.agents import AgentExecutor, create_react_agent

tools = [human_input]
llm = ChatOpenAI(model="gpt-4")

prompt = PromptTemplate.from_template("""你是一个有用的助手。
当你需要人类确认或额外信息时，使用 human_input 工具。

可用工具: {tools}
工具名称: {tool_names}

使用格式:
Question: 用户问题
Thought: 思考
Action: 工具名称
Action Input: 工具输入
Observation: 工具输出
... (重复)
Thought: 我知道答案了
Final Answer: 最终答案

开始!
Question: {input}
Thought: {agent_scratchpad}""")

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

result = agent_executor.invoke({
    "input": "帮我预订餐厅，需要确认人数和时间"
})

---

八、最佳实践

8.1 工具设计原则

原则	说明
单一职责	每个工具只做一件事
清晰描述	docstring 详细说明功能和参数
类型安全	使用类型注解和 Pydantic
错误处理	捕获异常，返回友好错误信息
幂等性	相同输入产生相同输出

8.2 安全最佳实践

from langchain_core.tools import tool
import os
import re

@tool
def safe_execute(command: str) -> str:
    """安全执行命令。
  
    Args:
        command: 要执行的命令
  
    Returns:
        执行结果
    """
    # 危险命令黑名单
    DANGEROUS_PATTERNS = [
        r"\brm\s+-[rf]",
        r"\b(format|mkfs)\b",
        r"\b(shutdown|reboot)\b",
        r"\b(sudo|su)\b",
        r"[;&|]",  # 命令链接
    ]
  
    for pattern in DANGEROUS_PATTERNS:
        if re.search(pattern, command):
            return f"错误: 命令被安全策略阻止"
  
    # 工作目录限制
    allowed_dir = os.environ.get("ALLOWED_DIR", "/tmp")
  
    try:
        result = os.popen(f"cd {allowed_dir} && {command}").read()
        return result
    except Exception as e:
        return f"执行错误: {e}"

8.3 性能优化

from langchain_core.tools import tool
from functools import lru_cache
import asyncio

# 1. 缓存工具结果
@tool
@lru_cache(maxsize=100)
def cached_search(query: str) -> str:
    """带缓存的搜索。"""
    # 相同查询直接返回缓存
    return f"搜索结果: {query}"

# 2. 并行执行多个工具
@tool
async def parallel_search(queries: str) -> str:
    """并行搜索多个查询。"""
    query_list = queries.split(",")
  
    async def search_one(q: str):
        await asyncio.sleep(0.1)  # 模拟 API 调用
        return f"{q}: 结果"
  
    results = await asyncio.gather(*[
        search_one(q.strip()) for q in query_list
    ])
    return "\n".join(results)

# 3. 超时控制
@tool
async def search_with_timeout(query: str) -> str:
    """带超时的搜索。"""
    try:
        async with asyncio.timeout(5):  # 5 秒超时
            await asyncio.sleep(0.1)
            return f"搜索结果: {query}"
    except asyncio.TimeoutError:
        return "搜索超时"

---

九、总结

9.1 Agent 架构图

9.2 选择指南

需求	推荐方案
简单工具调用	OpenAI Tools Agent
复杂推理	ReAct Agent
多步骤任务	Plan-and-Execute
自定义逻辑	继承 BaseSingleActionAgent
生产部署	LangGraph Agent

---

参考资料

• LangChain Agents 文档
• LangChain Tools 文档
• ReAct 论文
• OpenAI Function Calling