LangChain 智能体实战：Tool 定义、AgentExecutor 与 ReAct 循环

大模型在文本生成、摘要、翻译这类「处理语言」的任务上表现很强，但真实业务里大量任务并不止于语言：要查最新信息、要执行计算、要读数据库。LangChain 的智能体（Agent）功能解决的正是这个问题——它给模型配上一组「工具（Tool）」，让模型自己判断该在什么时候调用哪个工具，并把工具返回的结果接回推理过程，直到给出最终答案。这篇教程面向已经会写基础 Python、想动手搭一个可用智能体的开发者，从工具定义一路讲到执行器配置与内部循环。

如果你希望先了解站内相关的配套工具页，可以看 LangChain。

准备工作

在写第一行智能体代码之前，先把环境和概念对齐。

环境与依赖

需要 Python 运行环境，以及 LangChain 的核心包与所用模型供应商的集成包。典型安装方式如下：

pip install langchain langchain-core langchain-openai

如果打算接入其他模型供应商，把 langchain-openai 换成对应的集成包即可。各包的具体版本、支持的模型名称与接口变化较快，安装前请以官网当前信息为准。

API 密钥

调用托管模型需要一个可用的 API Key。代码里可以直接传参，但更稳妥的做法是通过环境变量注入，避免密钥写进版本库：

export OPENAI_API_KEY="your-api-key"

密钥的申请方式、可用地区、计费规则由各供应商决定，请以其官网当前信息为准。

三个核心概念

LangChain 的智能体由三部分构成，理解这三者的分工，后面看代码就不会迷路。

• LLM：负责思考与判断的语言模型，决定下一步做什么。
• Tool：模型可以调用的外部功能，比如搜索、计算、调 API、查库。
• AgentExecutor：管理整个执行循环的编排器，负责把模型的决策转成工具调用，再把结果喂回去。

智能体的运行流程是一个不断重复的循环：思考（Thought）→ 行动（Action）→ 观察（Observation）。这个模式通常被称为 ReAct（Reasoning + Acting）。

操作步骤

第一步：用装饰器定义工具

LangChain 提供了 @tool 装饰器，可以把一个普通 Python 函数直接注册成工具。下面定义三个最基础的工具：计算、取当前时间、模拟搜索。

from langchain.tools import tool

@tool
def calculate(expression: str) -> str:
    """数式を評価して結果を返します。例: '2 + 2' や '10 * 5'"""
    try:
        result = eval(expression)
        return f"計算結果: {result}"
    except Exception as e:
        return f"計算エラー: {str(e)}"

@tool
def get_current_time() -> str:
    """現在の日時を返します"""
    from datetime import datetime
    now = datetime.now()
    return f"現在時刻: {now.strftime('%Y-%m-%d %H:%M:%S')}"

@tool
def search_web(query: str) -> str:
    """Web検索を実行し、関連情報を返します"""
    # 実際の実装では検索API（SerpAPI、Tavilyなど）を呼び出す
    return f"'{query}' に関する検索結果: 最新情報が見つかりました"

每个工具都必须具备三个要素，缺一不可：

• 函数名：作为工具的标识符被模型引用。
• docstring：模型据此理解这个工具是干什么用的，这是最关键的一环。
• 类型标注：明确参数与返回值的类型。

模型是读着 docstring 来判断「什么时候该用哪个工具」的，所以描述要具体、直白，把适用场景和参数含义写清楚，不要写成一句含糊的占位说明。

第二步：构建智能体

工具就绪后，把它和模型、提示词模板组合起来。下面这段代码使用支持函数调用的模型来生成智能体。

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

# LLMの初期化
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0,
    api_key="your-api-key"
)

# Toolリスト
tools = [calculate, get_current_time, search_web]

# エージェント用プロンプトテンプレート
prompt = ChatPromptTemplate.from_messages([
    ("system", "あなたは有能なアシスタントです。与えられた道具を使ってユーザーの質問に答えてください。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

# エージェントの作成
agent = create_tool_calling_agent(llm, tools, prompt)

# AgentExecutorの初期化
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,              # 実行過程を詳細に出力
    max_iterations=5,          # 最大イテレーション数
    handle_parsing_errors=True
)

# エージェントの実行
result = agent_executor.invoke({
    "input": "現在の時刻を教えてください。また、15 × 8 の計算結果もお願いします。"
})
print(result["output"])

这里有几个要点值得单独说明。

提示词模板中的占位符：{input} 承接用户输入，{agent_scratchpad} 是留给中间推理过程的占位符，由执行器自动填充，不要手动替换成别的内容。

create_tool_calling_agent：它生成的是利用模型函数调用（Function Calling）能力的智能体，模型直接以结构化形式给出要调用的工具和参数，比纯文本解析更稳。

verbose=True：打开后，智能体的思考过程会打印到控制台，调试和观察行为时非常有用，生产环境可以关掉。

max_iterations：限制循环的最大轮数，防止模型在工具之间来回打转停不下来。

handle_parsing_errors：当模型输出无法被解析成合法动作时，把错误交回模型让它重试，而不是直接抛异常中断。

第三步：理解 ReAct 循环内部发生了什么

执行器内部跑的就是下面这个循环。用伪代码表示，逻辑大致是：

def react_agent_loop(user_input, tools, llm, max_iterations=5):
    scratchpad = []  # 過去の思考・行動・観察を記録
    for i in range(max_iterations):
        # 1. 思考（Thought）: 次に何をすべきか推論
        thought = llm.predict(
            f"ユーザー入力: {user_input}\n"
            f"過去のステップ: {scratchpad}\n"
            f"利用可能な道具: {[t.name for t in tools]}\n"
            f"次の思考: "
        )
        # 2. 行動（Action）: Toolを選択して実行
        action, action_input = parse_action(thought)
        if action == "finish":
            return generate_final_answer(scratchpad)
        tool = find_tool(action, tools)
        observation = tool.run(action_input)
        # 3. 観察（Observation）: 結果を記録
        scratchpad.append({
            "thought": thought,
            "action": action,
            "action_input": action_input,
            "observation": observation
        })
    return "最大イテレーション数に達しました"

实际使用 LangChain 时，这个循环由 AgentExecutor 自动完成，开发者不需要手动维护 scratchpad。但理解内部结构有实际好处：当提示词需要调整、或者要设计错误处理策略时，知道模型每一轮看到的是什么、上一轮的结果是怎么回填的，改起来才有方向。

第四步：编写更贴近实战的自定义工具

基础工具跑通之后，可以换成真实业务里会用到的工具。下面这组例子覆盖文件读取、数据库查询和写报告三个动作。

import os
import sqlite3
from typing import Optional
from langchain.tools import tool

@tool
def read_file_content(file_path: str) -> str:
    """指定されたファイルの内容を読み込んで返します"""
    try:
        if not os.path.exists(file_path):
            return f"エラー: ファイル '{file_path}' が見つかりません"
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read(1000)  # 最初の1000文字のみ読み込み
        if len(content) == 1000:
            content += "\n... (以下省略)"
        return f"ファイル内容:\n{content}"
    except Exception as e:
        return f"ファイル読み込みエラー: {str(e)}"

@tool
def search_database(query: str) -> str:
    """データベースから顧客情報を検索します。queryには顧客名またはIDを指定してください"""
    try:
        conn = sqlite3.connect("customers.db")
        cursor = conn.cursor()
        # SQLインジェクション対策としてパラメータ化クエリを使用
        cursor.execute(
            "SELECT name, email, plan FROM customers WHERE name LIKE ? OR id = ?",
            (f"%{query}%", query)
        )
        results = cursor.fetchall()
        conn.close()
        if not results:
            return f"'{query}' に一致する顧客は見つかりませんでした"
        output = "検索結果:\n"
        for row in results:
            output += f"- 名前: {row[0]}, Email: {row[1]}, プラン: {row[2]}\n"
        return output
    except Exception as e:
        return f"データベース検索エラー: {str(e)}"

@tool
def write_report(title: str, content: str) -> str:
    """レポートをファイルに書き込みます"""
    try:
        filename = f"report_{title.replace(' ', '_')}.txt"
        with open(filename, 'w', encoding='utf-8') as f:
            f.write(f"タイトル: {title}\n")
            f.write(f"作成日時: {__import__('datetime').datetime.now()}\n")
            f.write(f"\n{content}")
        return f"レポートを '{filename}' に保存しました"
    except Exception as e:
        return f"レポート作成エラー: {str(e)}"

把这几个工具挂到智能体上，就能处理复合指令。例如用户说「查一下客户 ID 12345 的信息，整理成一份报告」，智能体会自行完成这样的动作序列：先用 search_database 检索客户信息，再根据拿到的内容调用 write_report 生成文件，最后把结果回报给用户。整个过程不需要为这条指令写任何专门的分支逻辑。

一个完整示例

把前面的片段串起来，下面是一个从头到尾可以跑通的最小例子。它定义两个工具，构建智能体，并处理一个需要连续调用两次工具的问题。

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

@tool
def calculate(expression: str) -> str:
    """数式を評価して結果を返します。例: '2 + 2' や '10 * 5'"""
    try:
        return f"計算結果: {eval(expression)}"
    except Exception as e:
        return f"計算エラー: {str(e)}"

@tool
def get_current_time() -> str:
    """現在の日時を返します"""
    return f"現在時刻: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"

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

tools = [calculate, get_current_time]

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,
    max_iterations=5,
    handle_parsing_errors=True
)

result = agent_executor.invoke({
    "input": "現在の時刻を教えてください。また、15 × 8 の計算結果もお願いします。"
})

print(result["output"])

运行后，打开 verbose 会看到智能体先调用 get_current_time 拿到时间，再调用 calculate 算出 120，最后把两件事合并成一句回答。这就是 ReAct 循环在真实场景下的样子：一次用户输入，触发多轮工具调用。

注意事项

不要直接 eval 用户可控的表达式

前面示例里的 calculate 用了 eval，这在演示中够用，但放到生产环境是明确的安全风险——模型可能被诱导生成任意可执行表达式。更稳妥的做法是限定允许的运算符，用 AST 解析后自行求值：

import ast
import operator
from langchain.tools import tool

@tool
def safe_calculate(expression: str) -> str:
    """安全に数式を評価します"""
    ops = {
        ast.Add: operator.add,
        ast.Sub: operator.sub,
        ast.Mult: operator.mul,
        ast.Div: operator.truediv,
    }

    def _eval(node):
        if isinstance(node, ast.Num):
            return node.n
        elif isinstance(node, ast.BinOp):
            return ops[type(node.op)](_eval(node.left), _eval(node.right))
        else:
            raise ValueError("許可されていない式です")

    try:
        tree = ast.parse(expression, mode='eval')
        result = _eval(tree.body)
        return f"計算結果: {result}"
    except Exception as e:
        return f"計算エラー: {str(e)}"

同样的思路适用于所有工具：凡是会触碰文件系统、数据库、外部 API 的工具，都要假设输入是不可信的。数据库查询务必使用参数化查询，文件工具要限制可访问的路径范围。

关注 token 消耗与调用成本

智能体为了完成一个任务往往要发起多轮模型调用，token 消耗明显高于单次问答。LangChain 提供了回调来统计用量：

from langchain.callbacks import get_openai_callback

with get_openai_callback() as cb:
    result = agent_executor.invoke({"input": "複雑なタスク..."})

print(f"Total Tokens: {cb.total_tokens}")
print(f"Total Cost (USD): ${cb.total_cost:.4f}")

把 max_iterations 设成一个合理的值，既能防止无限循环，也能给单次任务的成本封顶。具体的计费单价与统计口径随供应商调整，请以官网当前信息为准。

其他容易踩的点

• 工具数量不是越多越好。工具越多，模型选择时的判断负担越重，docstring 之间的边界也越容易模糊。
• 工具函数内部要自己兜住异常并返回可读的错误文本，而不是让异常直接抛出。返回给模型的信息越清楚，它越有可能自行纠正。
• 涉及写文件、改数据这类有副作用的工具，建议在执行前加一层确认或权限校验，不要完全交给模型自主决定。
• 调试阶段保持 verbose=True，上线后关闭，避免把中间推理过程写进日志。

把工具定义、执行器配置和 ReAct 循环这三块吃透，智能体开发的基础就打牢了。真正决定一个智能体好不好用的，往往不是模型本身，而是工具边界划得清不清楚、错误信息给得够不够具体、循环上限设得合不合理。