第 3 课

Agent 与工具集成

LangChain for LLM App Dev — LangChain 应用开发

构建智能体、定义自定义工具、多工具协调。

Agent 与工具集成

课程简介

构建智能体、定义自定义工具、多工具协调。

🎬 本课程视频：LangChain for LLM App Dev — LangChain 应用开发

Agent 与工具集成

一、Agent 框架概述

LangChain 的 Agent 框架将 LLM 的推理能力、工具的执行能力和控制器的编排逻辑整合为完整的智能体系统。核心架构分为三层：

AgentExecutor（执行器）
    ↓ 控制循环
Agent（智能体）—— 负责推理和决策
    ↓ 调用工具
Tool（工具）—— 负责执行具体操作

二、Agent 执行器的控制循环

AgentExecutor 是运行 Agent 的核心引擎。它实现了一个「思考-行动-观察」循环：

from langchain.agents import AgentExecutor, create_react_agent
from langchain_openai import ChatOpenAI

# 初始化
llm = ChatOpenAI(model="gpt-4", temperature=0)

# AgentExecutor 的简化逻辑
class SimpleAgentExecutor:
    def __init__(self, agent, tools, max_iterations=5):
        self.agent = agent
        self.tools = {t.name: t for t in tools}
        self.max_iterations = max_iterations

    def invoke(self, input_data):
        # 初始化状态
        intermediate_steps = []

        for i in range(self.max_iterations):
            # 1. 调用 Agent 决定下一步
            output = self.agent.plan(intermediate_steps, input_data)

            # 2. 如果 Agent 决定给出最终答案，返回
            if output.is_final():
                return output.final_answer

            # 3. 执行工具调用
            tool = self.tools[output.tool_name]
            observation = tool.run(output.tool_input)

            # 4. 记录观察结果
            intermediate_steps.append((output, observation))

        # 达到最大迭代次数，强制返回
        return "达到最大迭代次数。"

关键参数：
- max_iterations：最大迭代次数，防止无限循环
- early_stopping_method：达到最大次数时的处理方式（generate / force）
- handle_parsing_errors：Agent 输出解析失败时的处理策略

三、工具定义：@tool 装饰器

3.1 基本用法

LangChain 提供了最简洁的方式来定义工具——@tool 装饰器：

from langchain.tools import tool
import requests

@tool
def get_weather(city: str) -> str:
    '''获取指定城市的当前天气信息。

    Args:
        city: 城市名称，如 北京、上海、深圳
    '''
    # 实际调用天气 API
    response = requests.get(f"https://api.weather.com/{city}")
    return response.text

@tool
def calculator(expression: str) -> str:
    '''执行数学计算。

    Args:
        expression: 数学表达式，如 (3 + 5) * 2
    '''
    try:
        return str(eval(expression))
    except Exception as e:
        return f"计算错误：{e}"

3.2 工具描述的重要性

@tool 装饰器会自动使用函数的文档字符串作为工具的描述。这个描述对 Agent 选择工具至关重要。

@tool
def search_database(query: str) -> str:
    '''在内部知识库中搜索相关信息。

    当你需要查询公司政策、产品信息、客户数据时使用此工具。
    输入关键词返回相关的知识库文章。
    不支持实时数据查询，如需查询实时数据请使用 search_web 工具。

    Args:
        query: 搜索关键词
    '''
    ...

好的描述应该包含：
1. 工具的功能：这个工具做什么
2. 使用场景：什么情况下应该使用这个工具
3. 限制条件：什么情况下不应该使用这个工具
4. 参数说明：每个参数的含义和格式

3.3 结构化工具

对于需要复杂参数的工具，可以使用 StructuredTool：

from langchain.tools import StructuredTool

def send_email(
    to: str,
    subject: str,
    body: str,
    cc: str = ""
) -> str:
    '''发送邮件。'''
    # 实现邮件发送逻辑
    return f"邮件已发送至 {to}"

email_tool = StructuredTool.from_function(
    func=send_email,
    name="send_email",
    description="发送邮件给指定收件人。"
)

四、多工具协调

4.1 工具选择策略

当 Agent 拥有多个工具时，它需要决定使用哪个工具。LangChain Agent 通过 ReAct 框架来实现：

# 定义多个工具
tools = [
    get_weather,
    calculator,
    search_database,
    send_email
]

# 创建 Agent
agent = create_react_agent(llm, tools, prompt)

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

result = agent_executor.invoke({
    "input": "北京今天天气怎么样？如果气温超过30度，发送邮件通知我"
})
# Agent 会：
# 1. 调用 get_weather("北京") 获取天气
# 2. 检查气温是否超过 30 度
# 3. 如果超过，调用 send_email 发送通知

4.2 工具共享上下文

有时工具之间需要共享信息。通过 AgentExecutor 的中间步骤，工具可以访问之前的执行结果：

# 在工具函数中，可以通过 context 参数访问中间步骤
@tool
def analyze_with_context(data: str, context: dict = None) -> str:
    '''基于已有数据进行进一步分析。'''
    if context and "previous_result" in context:
        # 使用之前工具的结果
        return f"基于之前的结果 {context['previous_result']}，分析得出：..."
    return f"分析 {data} 得出：..."

五、错误处理

5.1 工具调用失败

@tool
def fragile_api_call(param: str) -> str:
    '''可能失败的 API 调用。'''
    try:
        response = requests.get(f"https://api.example.com/{param}", timeout=5)
        response.raise_for_status()
        return response.text
    except requests.Timeout:
        return "TOOL_ERROR: API 超时，请稍后重试。"
    except requests.HTTPError as e:
        return f"TOOL_ERROR: API 返回错误 {e.response.status_code}"

AgentExecutor 提供了错误处理机制：
- max_iterations：防止无限重试
- handle_parsing_errors：解析失败时的处理
- 工具返回错误信息后，Agent 可以选择重试或尝试替代方案

5.2 错误恢复策略

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=10,
    max_execution_time=30,  # 最大执行时间 30 秒
    early_stopping_method="generate",  # 超时后生成部分答案
    handle_parsing_errors=True  # 解析出错时自动修复
)

5.3 回调机制

from langchain.callbacks import StdOutCallbackHandler

# 为 Agent 添加回调
handler = StdOutCallbackHandler()

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callbacks=[handler],  # 打印每个步骤的详细信息
    verbose=True
)

六、自定义 Agent

除了使用预定义的 Agent 类型，也可以创建自定义 Agent：

from langchain.agents import Agent
from langchain.schema import AgentAction, AgentFinish

class CustomAgent(Agent):
    '''自定义 Agent'''

    @property
    def observation_prefix(self) -> str:
        return "观察："

    @property
    def llm_prefix(self) -> str:
        return "思考："

    def _get_tools(self) -> list:
        return self.allowed_tools

七、最佳实践

清晰命名工具：工具名称应该直观反映其功能（search_web > s_web）
详细描述工具：描述中明确使用场景和限制，帮助 Agent 正确选择
设置最大迭代次数：防止 Agent 陷入无限循环
错误处理：工具需要处理异常情况，返回清晰的错误信息
监控和日志：记录 Agent 的思考链条和工具调用记录，用于调试和优化

八、总结

LangChain 的 Agent 框架提供了构建智能体系统的完整工具集。@tool 装饰器让工具定义简洁直观，AgentExecutor 实现了可配置的思考-行动-观察循环，多工具协调模式让 Agent 能够组合使用多个工具完成复杂任务。清晰的工具定义和鲁棒的错误处理是构建可靠生产级 Agent 系统的核心要素。

七、多工具协调模式

7.1 工具选择策略

当 Agent 有多个工具可用时，如何选择正确的工具？

名称匹配：工具名称与任务描述匹配
描述匹配：LLM 依据工具描述选择合适的工具
参数匹配：LLM 根据任务的参数生成工具调用参数

7.2 工具协调的常见模式

链式工具调用：一个工具的輸出作为另一个工具的输入

搜索(查询="Python 异步编程") → 摘要(搜索结果) → 翻译(摘要, 目标语言="中文")

并行工具调用：同时调用多个独立的工具

搜索(查询="特斯拉股价")
搜索(查询="苹果股价")  # 同时执行

条件工具调用：根据条件决定调用哪个工具

if 需要实时数据: 调用 search API
elif 需要计算: 调用 calculator
else: 直接回答

八、总结

关键要点回顾：
- Agent 框架通过 ReAct 循环实现自主推理和行动
- @tool 装饰器让工具定义更简洁
- “思考→行动→观察”循环是 Agent 的核心
- 工具描述的质量决定工具选择准确率
- 多工具协调需要设计合理的调用策略

七、多工具协调模式

7.1 工具选择策略

当 Agent 有多个工具可用时，如何选择正确的工具？

名称匹配：工具名称与任务描述匹配
描述匹配：LLM 依据工具描述选择合适的工具
参数匹配：LLM 根据任务的参数生成工具调用参数

7.2 工具协调的常见模式

链式工具调用：一个工具的輸出作为另一个工具的输入

搜索(查询="Python 异步编程") → 摘要(搜索结果) → 翻译(摘要, 目标语言="中文")

并行工具调用：同时调用多个独立的工具

搜索(查询="特斯拉股价")
搜索(查询="苹果股价")  # 同时执行

条件工具调用：根据条件决定调用哪个工具

if 需要实时数据: 调用 search API
elif 需要计算: 调用 calculator
else: 直接回答

八、总结

七、多工具协调模式

7.1 工具选择策略

当 Agent 有多个工具可用时，如何选择正确的工具？

名称匹配：工具名称与任务描述匹配
描述匹配：LLM 依据工具描述选择合适的工具
参数匹配：LLM 根据任务的参数生成工具调用参数

7.2 工具协调的常见模式

链式工具调用：一个工具的輸出作为另一个工具的输入

搜索(查询="Python 异步编程") → 摘要(搜索结果) → 翻译(摘要, 目标语言="中文")

并行工具调用：同时调用多个独立的工具

搜索(查询="特斯拉股价")
搜索(查询="苹果股价")  # 同时执行

条件工具调用：根据条件决定调用哪个工具

if 需要实时数据: 调用 search API
elif 需要计算: 调用 calculator
else: 直接回答

Agent 与工具集成

Agent 与工具集成

课程简介

Agent 与工具集成

一、Agent 框架概述

二、Agent 执行器的控制循环

三、工具定义：@tool 装饰器

3.1 基本用法

3.2 工具描述的重要性

3.3 结构化工具

四、多工具协调

4.1 工具选择策略

4.2 工具共享上下文

五、错误处理

5.1 工具调用失败

5.2 错误恢复策略

5.3 回调机制

六、自定义 Agent

七、最佳实践

八、总结

七、多工具协调模式

7.1 工具选择策略

7.2 工具协调的常见模式

八、总结

七、多工具协调模式

7.1 工具选择策略

7.2 工具协调的常见模式

八、总结

七、多工具协调模式

7.1 工具选择策略

7.2 工具协调的常见模式

八、总结

延伸阅读