我们在解决的问题

你有一个能完美执行指令的语言模型——它能写代码、回答问题、推理问题。但每个开发者都会遇到一堵墙：仅靠 LLM 实际上无法在真实世界中做任何事。

你的模型能告诉你天气如何，但它无法调用天气 API。它能解释如何写数据库查询，但它无法对你的实际数据库执行查询。它能建议使用哪个工具，但需要你来编写逻辑才能实际调用那个工具。

这个差距——LLM 能决定做什么和实际能执行之间的差距——正是 LangChain 智能体要填补的。它们做得优雅：描述你的工具，智能体决定何时使用它们，框架处理所有底层连接。

但"智能体"不是一件事。它是一整套技术栈。在你能构建一个与 API 对话、处理错误、并有人工反馈循环的生产级智能体之前，你需要理解 Runnables——正是这个基本构建块让一切运转起来。

本指南将带你走完整条技术栈。阅读完毕后，你不仅会理解如何构建智能体，还会理解每个部分为什么存在，以及如何为生产工作负载组合它们。

• Runnables：基础——理解驱动一切的接口
• 链式组件——构建你的第一个多步骤工作流
• 构建代码生成器——一个将 LLM 调用链式组合的实用第一个智能体
• 工具：教智能体行动——如何赋予你的智能体超能力
• 创建自定义工具——包装你的 Python 代码以便智能体使用
• 工具绑定与调用——模型如何决定使用哪个工具
• 智能体循环：执行与控制——手动构建智能体以获得细粒度控制
• 集成真实 API——一个完整的天气智能体示例
• 高级模式与 create_agent——现代 LangChain 快捷方式及使用场景
• 陷阱与生产经验——常见失败及如何避免

1. Runnables：基础

在智能体之前，在工具之前，在所有魔法之前——只有 Runnable。

如果你用过 LangChain，你可能见过这种模式：

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4")
result = model.invoke("What is 2 + 2?")

那个 invoke() 调用？就是 Runnable。模型是一个 Runnable。理解 Runnable 是什么，是理解后续一切的关键。

Runnable 是 LangChain 对任何接受输入、处理并返回输出的事物的抽象。它是一种契约："我可以被调用、流式处理、批处理，或与其他事物组合。"

可以这样理解：如果你用过 Unix 管道（cat file | grep pattern | wc -l），Runnable 的工作方式相同。每个管道操作是独立的，但你可以组合它们。每一步不需要知道其他步骤——只需理解输入和输出。

核心洞见：LangChain 中的一切都是 Runnable。模型、提示模板、输出解析器、工具、链——它们都实现了相同的接口。这正是组合如此强大的原因。

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 下面每个都是 Runnable
prompt = ChatPromptTemplate.from_template("Tell me a joke about {topic}")
model = ChatOpenAI(model="gpt-4")
parser = StrOutputParser()

# 用管道操作符 | 将它们链式组合
# 这会创建一个新的 Runnable，包含这三个组件
joke_chain = prompt | model | parser

# 现在用输入调用它
result = joke_chain.invoke({"topic": "debugging"})
print(result)
# 输出: "Why do programmers prefer dark mode? Because light attracts bugs!"

注意 | 操作符。这就是管道——它将 Runnable 链接在一起。当你调用最终链时，LangChain 自动完成：

1. 将输入传递给第一个 Runnable（prompt）
2. 将输出传递给下一个（model）
3. 将输出传递给下一个（parser）
4. 返回最终结果

这是声明式组合。你描述流程，LangChain 处理线程逻辑。

为什么这对智能体很重要

智能体只是更复杂的 Runnable 链。一个智能体是：

• 一个工具执行器（Runnable）
• 循环执行直到模型说"停止"

一旦你掌握了 Runnable 模式，智能体就不再神秘了。它们只是组合的进一步延伸。

2. 链式组件

让我们构建稍微复杂一点的东西——一个组合多个步骤的链。

假设你要构建一个产品推荐系统。你需要：

1. 获取用户偏好
2. 将其传递给 LLM 生成搜索查询
3. 用该查询查找产品
4. 将结果传回 LLM 编写推荐

以下是使用 Runnable 构建的方式：

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 步骤 1：从偏好生成搜索查询
query_prompt = ChatPromptTemplate.from_template(
    """Given these user preferences, generate a concise search query.

User preferences: {preferences}

Search query:"""
)

model = ChatOpenAI(model="gpt-4")
query_parser = StrOutputParser()

# 这个链将偏好 → 搜索查询
search_query_chain = query_prompt | model | query_parser

# 步骤 2：模拟产品搜索（实际中调用 API）
def search_products(query):
    """模拟产品搜索。生产环境中会查询你的数据库。"""
    # 假设我们找到了这些产品
    return f"Found products matching '{query}': Laptop Pro 15, Gaming Mouse RGB, USB-C Hub"

# 步骤 3：从搜索结果生成推荐
recommendation_prompt = ChatPromptTemplate.from_template(
    """You are a product recommendation expert.

User preferences: {preferences}
Search results: {search_results}

Write a brief recommendation based on these results."""
)

recommendation_chain = recommendation_prompt | model | query_parser

# 现在组合整个流程
from langchain_core.runnables import RunnablePassthrough

# RunnablePassthrough 让原始输入流经整个链
full_chain = (
    RunnablePassthrough.assign(search_results=search_query_chain | (lambda q: search_products(q)))
    | recommendation_chain
)

# 执行它
result = full_chain.invoke({"preferences": "I need a laptop for video editing"})
print(result)

这仍然只是一个链——确定性的，没有循环。但注意这个模式：你描述结构，组合 Runnable，然后调用整个流程。智能体遵循完全相同的模式，只是增加了循环和动态选择使用哪个工具的能力。

3. 构建代码生成器：一个实用第一个智能体

让我们构建一个感觉更像智能体的东西：一个为用户请求生成 Python 代码，并可选地改进它的链。

这还不是完整的智能体（没有动态工具选择），但它展示了真实智能体中你会看到的链式模式。

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

model = ChatOpenAI(model="gpt-4")

# 步骤 1：生成初始代码
code_gen_prompt = ChatPromptTemplate.from_template(
    """You are an expert Python developer.
Generate clean, well-commented Python code for this request.

Request: {request}

Only output the code, no explanations."""
)

# 步骤 2：审查和改进代码
review_prompt = ChatPromptTemplate.from_template(
    """Review this Python code for:
- Efficiency
- Readability
- Best practices
- Potential bugs

Code:
{code}

Provide the improved code with inline comments explaining any changes."""
)

# 组合成管道
code_gen_chain = code_gen_prompt | model | StrOutputParser()

# 关键在这里：在传递原始请求的同时，计算生成的代码
code_review_chain = (
    RunnablePassthrough.assign(code=code_gen_chain)
    | review_prompt
    | model
    | StrOutputParser()
)

# 执行
result = code_review_chain.invoke({
    "request": "Write a function that checks if a number is prime"
})

print(result)

底层发生的事情：

• RunnablePassthrough.assign() 保留原始输入并添加一个新字段（code），通过运行 code_gen_chain 计算得出
• 丰富的输入流向 review_prompt
• 模型审查代码
• 解析器返回纯文本

这个模式对智能体至关重要：在每一步，你都在用新信息丰富上下文，那个上下文流向下一步。

4. 工具：教智能体行动

现在我们进入真正的力量。Tool 是你告诉智能体："这是你可以在真实世界中做的事情。"

在构建自定义工具之前，让我们理解 Tool 是什么：

from langchain_core.tools import tool

使用 @tool 装饰器定义工具

@tool
def get_current_time() -> str:
    """Get the current date and time.

This tool is useful when the user asks about the current time,
    date, or needs to schedule something."""
    from datetime import datetime
    return datetime.now().isoformat()

@tool
def calculator(expression: str) -> str:
    """Evaluate a mathematical expression.

Args:
        expression: A valid Python math expression (e.g., '2 + 2' or 'sqrt(16)')

This tool is useful for calculations the model should perform precisely."""
    try:
        # Use Python's eval (never do this in production with untrusted input!)
        result = eval(expression)
        return f"Result: {result}"
    except Exception as e:
        return f"Error: {e}"

# Tools are now Runnables
print(get_current_time.invoke({}))
print(calculator.invoke({"expression": "2 ** 10"}))

注意看这些文档字符串。文档字符串是 LLM 看到的内容，它相当于你给模型的 API 文档。好的文档字符串 = 好的工具调用。

• 一个名称（从函数名派生）
• 一个描述（文档字符串）
• 输入模式（函数参数）
• 一个可调用的 Runnable

模型永远不会看到你的代码，它只看到工具名称、描述和参数名称。所以写文档字符串时要像在向一个非程序员解释一样。

5. 创建自定义工具

让我们为一个实际场景构建工具：一个辅助数据分析的助手。

from langchain_core.tools import tool
import json

# 工具 1：解析 CSV 类数据
@tool
def parse_csv_data(data_str: str) -> str:
    """Parse a CSV string into structured data.

Args:
        data_str: CSV data as a string (comma-separated values)

Returns JSON representation of the data with column headers and rows.

Example: If given "name,age\nAlice,30\nBob,25", returns a JSON array.

Use this when you need to work with CSV data."""
    try:
        lines = data_str.strip().split('\n')
        headers = lines[0].split(',')
        rows = []
        for line in lines[1:]:
            values = line.split(',')
            row = {headers[i]: values[i] for i in range(len(headers))}
            rows.append(row)
        return json.dumps(rows, indent=2)
    except Exception as e:
        return f"Error parsing CSV: {e}"

# 工具 2：计算统计量
@tool
def calculate_statistics(numbers_str: str) -> str:
    """Calculate basic statistics (mean, median, std dev, min, max) from a list of numbers.

Args:
        numbers_str: Space or comma-separated numbers

Use this when the user asks for statistics or summary data."""
    import statistics
    try:
        # Handle both space and comma separation
        numbers = [float(x) for x in numbers_str.replace(',', ' ').split()]

        return json.dumps({
            "count": len(numbers),
            "mean": statistics.mean(numbers),
            "median": statistics.median(numbers),
            "stdev": statistics.stdev(numbers) if len(numbers) > 1 else 0,
            "min": min(numbers),
            "max": max(numbers)
        }, indent=2)
    except Exception as e:
        return f"Error calculating statistics: {e}"

# 工具 3：过滤数据
@tool
def filter_data(json_data: str, field: str, value: str) -> str:
    """Filter a JSON array by matching a field to a value.

Args:
        json_data: JSON array as a string
        field: The field name to filter by
        value: The value to match

Returns JSON array containing only matching rows.

Example: Filter users by age == 30."""
    try:
        data = json.loads(json_data)
        filtered = [row for row in data if str(row.get(field)) == value]
        return json.dumps(filtered, indent=2)
    except Exception as e:
        return f"Error filtering data: {e}"

# 收集工具供 Agent 使用
tools = [parse_csv_data, calculate_statistics, filter_data]

自定义工具的关键原则：

• 文档字符串就是你的 API 契约 — 模型只能看到名称和文档字符串，看不到代码
• 保持简单 — 工具应该做好一件事，而不是成为一个小型应用程序
• 优雅地处理错误 — 返回有意义的错误信息，而不是堆栈跟踪
• 独立测试 — 在将工具添加到 Agent 之前，先验证它能独立运行
• 类型提示很重要 — 它们会成为模型看到的工具模式的一部分

6. 工具绑定与调用

这才是有趣的地方。模型怎么知道要调用工具？我们又如何实际执行它？

LangChain 利用现代 LLM 的函数调用能力，流程如下：

1. 你给模型一组工具
2. 用户提问
3. 模型看到用户的请求和可用工具列表
4. 模型决定：「我应该用这些参数调用工具 X」
5. LangChain 拦截这个决定并实际调用工具
6. 结果返回给模型
7. 模型根据工具结果生成响应

让我们看实际用法：

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain.tools.render import format_tool_to_openai_function_calls

model = ChatOpenAI(model="gpt-4")

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers together."""
    return a * b

@tool
def add(a: int, b: int) -> int:
    """Add two numbers together."""
    return a + b

tools = [multiply, add]

# Bind tools to the model
# This tells the model: "You have these tools available, and here's how to request them"
model_with_tools = model.bind_tools(tools)

# When we invoke it, the model returns a message with a "tool_call" in it
response = model_with_tools.invoke("What is 5 times 3, plus 7?")

print("Response:", response)
print("Tool calls:", response.tool_calls)

# The response includes:
# - .content: The text response
# - .tool_calls: A list of tool invocations the model decided to make
# Example tool_call:
# {
#     "name": "multiply",
#     "args": {"a": 5, "b": 3},
#     "id": "call_123"
# }

关键时刻：当你调用 invoke() 时，模型会立即返回其工具决策，它不会执行工具。那是你的工作（或者说，是接下来要介绍的 Agent 循环的工作）。

模型在说：「这是我想做的下一步。」你（程序员）决定是否真的要执行它、是否要做错误检查，或者是否要根据人类反馈修改它。

7. Agent 循环：手动构建

Agent 循环处理的是模型和工具之间的来回交互：模型决定调用工具，你执行它，把结果反馈给模型，不断重复，直到模型说「完成」。

以下是手动 Agent 循环 — 这是更高层抽象隐藏的本质：

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage, AIMessage

# 定义工具
@tool
def search_knowledge_base(query: str) -> str:
    """Search the company knowledge base for information.

Args:
        query: A search query

Returns relevant documents."""
    # Mock implementation
    if "python" in query.lower():
        return "Document 1: Python Best Practices\nDocument 2: Python Performance Tips"
    return "No documents found"

@tool
def fetch_docs(doc_id: str) -> str:
    """Fetch the full content of a document by ID.

Args:
        doc_id: The document identifier

Returns the full document content."""
    docs = {
        "doc1": "Python is a high-level language...",
        "doc2": "Use list comprehensions for performance...",
    }
    return docs.get(doc_id, "Document not found")

tools = [search_knowledge_base, fetch_docs]
model = ChatOpenAI(model="gpt-4")
model_with_tools = model.bind_tools(tools)

# Agent 循环
def run_agent_loop(user_input: str, max_iterations: int = 10):
    """
    Run the agent loop until the model stops requesting tools.

This manually handles:
    - Sending user input to the model
    - Detecting tool calls in the response
    - Executing tools
    - Feeding results back to the model
    - Looping until the model says "done"
    """

    # Start with the user's message
    messages = [HumanMessage(content=user_input)]

    for i in range(max_iterations):
        # Get the model's response (might include tool calls)
        response = model_with_tools.invoke(messages)

        # If the model didn't request any tools, it's done
        if not response.tool_calls:
            return response.content

        # Add the model's response to the message history
        # This is important: the model needs to see its own reasoning
        messages.append(AIMessage(content=response.content, tool_calls=response.tool_calls))

        # Execute each tool call the model requested
        for tool_call in response.tool_calls:
            tool_name = tool_call['name']
            tool_args = tool_call['args']

            # Find the tool and execute it
            tool = next((t for t in tools if t.name == tool_name), None)
            if not tool:
                result = "Error: Tool not found"
            else:
                result = tool.invoke(tool_args)

# 将工具结果添加到消息历史中

```python
# The model will see what the tool returned
messages.append(ToolMessage(
    content=result,
    tool_call_id=tool_call['id']
))

print(f"[Iteration {i+1}] Agent decided to call tools. Processing...")

return "Max iterations reached"

Run it

result = run_agent_loop("I need to learn about Python performance optimization") print("Final answer:", result)

以下是整个流程：

**初始输入**：用户消息进入消息列表

**模型调用**：使用迄今为止的所有消息调用模型

**工具决策**：检查模型是否决定使用任何工具

**工具执行**：实际运行工具并捕获结果

**反馈循环**：将工具结果添加回消息，使模型能够看到发生了什么

**重复**：下一次迭代时，模型拥有完整的上下文，包括之前的工具结果

消息历史是关键。每一次迭代，你都在构建一份对话记录：

Human: [问题] AI: 我来搜索一下信息... [tool_call: search] Tool: [搜索结果] AI: 现在我来获取完整文档... [tool_call: fetch] Tool: [文档内容] AI: 根据这些文档，以下是我发现的内容...

这个手动循环让你理解正在发生的事情。在生产环境中，你会使用 LangChain 的 AgentExecutor 或 create_agent，它们会为你处理这个循环。但理解手动循环至关重要——这是 bug 出现的地方，是需要人工介入的地方，也是你可以添加监控的地方。

## 8. 集成真实 API：一个完整的天气智能体

让我们构建一个实用的天气智能体，它可以：

- 查询某个城市的当前天气
- 根据天气建议活动

```python
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
import requests
import json

# 使用 OpenWeatherMap API 的真实工具
@tool
def get_current_weather(city: str) -> str:
    """Get the current weather for a city.

Args:
        city: The city name (e.g., 'London', 'Tokyo')

Returns weather information including temperature, conditions, and humidity.

Use this when the user asks about current weather."""
    try:
        # In production, use your actual API key
        api_key = "YOUR_OPENWEATHER_API_KEY"
        url = f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}&units=metric"

        response = requests.get(url, timeout=5)
        response.raise_for_status()
        data = response.json()

        return json.dumps({
            "city": data.get('name'),
            "temp": data['main']['temp'],
            "feels_like": data['main']['feels_like'],
            "condition": data['weather'][0]['main'],
            "humidity": data['main']['humidity'],
            "wind_speed": data['wind']['speed']
        })
    except requests.RequestException as e:
        return f"Error fetching weather: {e}"

@tool
def get_weather_forecast(city: str, days: int = 3) -> str:
    """Get weather forecast for a city.

Args:
        city: The city name
        days: Number of days to forecast (1-5)

Returns forecast data."""
    try:
        api_key = "YOUR_OPENWEATHER_API_KEY"
        # Using free tier endpoint
        url = f"https://api.openweathermap.org/data/2.5/forecast?q={city}&appid={api_key}&units=metric"
        response = requests.get(url, timeout=5)
        response.raise_for_status()
        data = response.json()

        # Parse into forecast summary
        forecasts = []
        for item in data['list'][::8]:  # Every 8 entries = ~1 day
            forecasts.append({
                "time": item['dt_txt'],
                "temp": item['main']['temp'],
                "condition": item['weather'][0]['main']
            })

        return json.dumps(forecasts[:days])
    except requests.RequestException as e:
        return f"Error fetching forecast: {e}"

@tool
def suggest_activity(weather_condition: str, temperature: float) -> str:
    """Suggest activities based on weather conditions.

Args:
        weather_condition: Weather type (e.g., 'Sunny', 'Rainy', 'Cloudy')
        temperature: Temperature in Celsius

Returns activity suggestions appropriate for the weather."""
    suggestions = {
        "Sunny": {
            "hot": "🏖️ Beach, outdoor sports, cycling",
            "warm": "⛳ Golf, hiking, picnic",
            "cool": "🚴 Jogging, sightseeing"
        },
        "Rainy": {
            "hot": "🎬 Indoor activities, museum, shopping",
            "warm": "📚 Reading, indoor sports",
            "cool": "☕ Cozy cafes, bookstores"
        },
        "Cloudy": {
            "hot": "🎮 Outdoor games, park",
            "warm": "🎨 Photography, exploring",
            "cool": "🥾 Hiking, nature walks"
        }
    }

    # Categorize temperature
    temp_cat = "hot" if temperature > 25 else ("warm" if temperature > 15 else "cool")
    weather_suggestions = suggestions.get(weather_condition, suggestions["Cloudy"])

    return weather_suggestions.get(temp_cat, "Indoor activities recommended")

tools = [get_current_weather, get_weather_forecast, suggest_activity]

现在让我们构建这个智能体：

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, ToolMessage, AIMessage

model = ChatOpenAI(model="gpt-4")
model_with_tools = model.bind_tools(tools)

def run_weather_agent(user_input: str, max_iterations: int = 10):
    """Run the weather agent loop."""
    messages = [HumanMessage(content=user_input)]

    for iteration in range(max_iterations):
        response = model_with_tools.invoke(messages)

        # If no tool calls, the model is done
        if not response.tool_calls:
            print(f"\n✅ Agent response:\n{response.content}")
            return response.content

        # Add model's message with tool calls
        messages.append(AIMessage(content=response.content, tool_calls=response.tool_calls))

        # Execute each tool
        for tool_call in response.tool_calls:
            tool_name