什么是 LangChain？

LangChain 是构建由大型语言模型（LLM）驱动的应用程序和智能体的最便捷方式。通过不到 10 行代码，您可以连接到 OpenAI、Anthropic、Google 等主流模型提供商。

v1.0 版本核心特性

LangChain v1.0 是一个重要的里程碑版本，带来了以下核心改进：

1. 标准化的模型接口：统一不同提供商的 API，实现无缝切换
2. 易于使用的智能体：10 行代码即可创建功能强大的智能体
3. 基于 LangGraph 构建：享受持久化执行、人机交互、流式处理等高级特性
4. 强大的调试能力：通过 LangSmith 获得深度可视化和追踪能力

LangChain vs LangGraph

• LangChain：适合快速构建智能体和自主应用，提供预构建的智能体架构
• LangGraph：低级别的智能体编排框架，适合需要确定性工作流、深度定制和精确控制延迟的高级场景

---

环境安装与配置

基础安装

# 安装 LangChain v1.0（需要 Python 3.10+）
pip install -U langchain

# 安装特定模型提供商的集成（按需选择）
pip install -qU "langchain[anthropic]"  # Anthropic Claude
pip install -qU "langchain[openai]"     # OpenAI GPT
pip install -qU "langchain[google]"     # Google Gemini

环境变量配置

创建 .env 文件或在系统环境变量中设置：

# OpenAI
export OPENAI_API_KEY="your-openai-api-key"

# Anthropic
export ANTHROPIC_API_KEY="your-anthropic-api-key"

# Google
export GOOGLE_API_KEY="your-google-api-key"

# LangSmith（可选，用于调试和追踪）
export LANGCHAIN_TRACING_V2="true"
export LANGCHAIN_API_KEY="your-langsmith-api-key"

验证安装

import langchain
print(f"LangChain 版本: {langchain.__version__}")

---

模型使用

1. 基础模型调用

使用 OpenAI 模型

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

# 创建模型实例
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0.7,
    max_tokens=1000
)

# 单次调用
response = llm.invoke("什么是人工智能？")
print(response.content)

使用 Anthropic Claude 模型

from langchain_anthropic import ChatAnthropic

# 创建 Claude 模型实例
llm = ChatAnthropic(
    model="claude-sonnet-4-5-20250929",
    temperature=0.7
)

# 调用模型
response = llm.invoke("解释一下量子计算的基本原理")
print(response.content)

使用 Google Gemini 模型

from langchain_google_genai import ChatGoogleGenerativeAI

# 创建 Gemini 模型实例
llm = ChatGoogleGenerativeAI(
    model="gemini-pro",
    temperature=0.7
)

response = llm.invoke("Python 中如何实现异步编程？")
print(response.content)

2. 流式响应

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", streaming=True)

# 流式调用
for chunk in llm.stream("写一首关于春天的诗"):
    if chunk.content:
        print(chunk.content, end="", flush=True)

3. 批量调用

from langchain_openai import ChatOpenAI

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

# 批量处理多个输入
messages = [
    [HumanMessage(content="什么是机器学习？")],
    [HumanMessage(content="什么是深度学习？")],
    [HumanMessage(content="什么是强化学习？")]
]

responses = llm.batch(messages)
for response in responses:
    print(response.content)

4. 带系统提示词的对话

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage

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

messages = [
    SystemMessage(content="你是一个专业的技术顾问，擅长解释复杂的技术概念。"),
    HumanMessage(content="请用简单的语言解释什么是区块链？")
]

response = llm.invoke(messages)
print(response.content)

---

工具（Tools）开发

工具是智能体与外部世界交互的接口。LangChain 支持多种工具类型。

1. 基础函数工具

最简单的工具就是 Python 函数：

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息。
    
    Args:
        city: 城市名称，例如：北京、上海、深圳
    """
    # 这里应该是实际的天气 API 调用
    # 示例返回
    weather_data = {
        "北京": "晴天，温度 25°C",
        "上海": "多云，温度 22°C",
        "深圳": "小雨，温度 28°C"
    }
    return weather_data.get(city, f"{city}的天气信息暂时无法获取")

# 使用工具
result = get_weather.invoke({"city": "北京"})
print(result)

2. 计算器工具

from langchain_core.tools import tool
import math

@tool
def calculator(expression: str) -> str:
    """执行数学计算。
    
    Args:
        expression: 数学表达式，例如：2+2, 10*5, sqrt(16)
    """
    try:
        # 安全的数学计算
        result = eval(expression, {"__builtins__": {}}, {
            "sqrt": math.sqrt,
            "sin": math.sin,
            "cos": math.cos,
            "tan": math.tan,
            "pi": math.pi,
            "e": math.e
        })
        return str(result)
    except Exception as e:
        return f"计算错误: {str(e)}"

# 使用示例
result = calculator.invoke({"expression": "sqrt(16) + 10*2"})
print(result)

3. 网络搜索工具

from langchain_core.tools import tool
import requests

@tool
def search_web(query: str) -> str:
    """在网络上搜索信息。
    
    Args:
        query: 搜索关键词
    """
    # 这里应该调用实际的搜索 API（如 Google Search API、Serper API 等）
    # 示例实现
    try:
        # 实际项目中应该使用真实的搜索 API
        return f"关于'{query}'的搜索结果：这是一个示例返回，实际应该调用搜索 API"
    except Exception as e:
        return f"搜索失败: {str(e)}"

4. 数据库查询工具

from langchain_core.tools import tool
import sqlite3

@tool
def query_database(query: str) -> str:
    """执行数据库查询。
    
    Args:
        query: SQL 查询语句
    """
    try:
        # 示例：SQLite 数据库
        conn = sqlite3.connect("example.db")
        cursor = conn.cursor()
        cursor.execute(query)
        results = cursor.fetchall()
        conn.close()
        return str(results)
    except Exception as e:
        return f"查询错误: {str(e)}"

5. 文件操作工具

from langchain_core.tools import tool
import os

@tool
def read_file(file_path: str) -> str:
    """读取文件内容。
    
    Args:
        file_path: 文件路径
    """
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            return f.read()
    except Exception as e:
        return f"读取文件失败: {str(e)}"

@tool
def write_file(file_path: str, content: str) -> str:
    """写入文件内容。
    
    Args:
        file_path: 文件路径
        content: 要写入的内容
    """
    try:
        with open(file_path, 'w', encoding='utf-8') as f:
            f.write(content)
        return f"文件已成功写入: {file_path}"
    except Exception as e:
        return f"写入文件失败: {str(e)}"

6. 工具组合示例

from langchain_core.tools import tool

# 定义多个工具
@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息。"""
    return f"{city}的天气：晴天，25°C"

@tool
def get_time(timezone: str = "UTC") -> str:
    """获取当前时间。
    
    Args:
        timezone: 时区，默认为 UTC
    """
    from datetime import datetime
    return f"当前时间（{timezone}）: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"

@tool
def translate_text(text: str, target_language: str) -> str:
    """翻译文本到目标语言。
    
    Args:
        text: 要翻译的文本
        target_language: 目标语言代码，如：en, zh, ja
    """
    # 这里应该调用实际的翻译 API
    return f"[{target_language}] {text}"

# 工具列表
tools = [get_weather, get_time, translate_text]

---

智能体（Agent）创建与使用

1. 最简单的智能体

根据官方文档，创建智能体只需要几行代码：

from langchain.agents import create_agent
from langchain_anthropic import ChatAnthropic

# 定义工具
def get_weather(city: str) -> str:
    """获取指定城市的天气信息。"""
    return f"{city}的天气：晴天，25°C"

# 创建智能体
agent = create_agent(
    model="claude-sonnet-4-5-20250929",
    tools=[get_weather],
    system_prompt="你是一个有用的助手",
)

# 运行智能体
response = agent.invoke(
    {"messages": [{"role": "user", "content": "北京的天气怎么样？"}]}
)

print(response["messages"][-1].content)

2. 使用 OpenAI 模型的智能体

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

# 定义多个工具
@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息。"""
    return f"{city}的天气：晴天，25°C"

@tool
def calculator(expression: str) -> str:
    """执行数学计算。"""
    try:
        result = eval(expression)
        return str(result)
    except:
        return "计算错误"

# 创建智能体
agent = create_agent(
    model=ChatOpenAI(model="gpt-4o", temperature=0),
    tools=[get_weather, calculator],
    system_prompt="你是一个智能助手，可以帮助用户查询天气和执行计算。",
)

# 运行智能体
response = agent.invoke({
    "messages": [{
        "role": "user", 
        "content": "北京天气如何？另外帮我计算 123 * 456 等于多少？"
    }]
})

print(response["messages"][-1].content)

3. 流式智能体响应

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息。"""
    return f"{city}的天气：晴天，25°C"

agent = create_agent(
    model=ChatOpenAI(model="gpt-4o", streaming=True),
    tools=[get_weather],
    system_prompt="你是一个有用的助手",
)

# 流式调用
for chunk in agent.stream({
    "messages": [{"role": "user", "content": "告诉我北京、上海、深圳的天气"}]
}):
    if "messages" in chunk:
        for message in chunk["messages"]:
            if hasattr(message, "content") and message.content:
                print(message.content, end="", flush=True)

4. 带记忆的智能体

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息。"""
    return f"{city}的天气：晴天，25°C"

agent = create_agent(
    model=ChatOpenAI(model="gpt-4o"),
    tools=[get_weather],
    system_prompt="你是一个有用的助手，会记住之前的对话内容。",
)

# 第一轮对话
response1 = agent.invoke({
    "messages": [{"role": "user", "content": "我的名字是张三"}]
})
print("助手:", response1["messages"][-1].content)

# 第二轮对话（智能体会记住名字）
response2 = agent.invoke({
    "messages": response1["messages"] + [{"role": "user", "content": "我刚才告诉你我的名字是什么？"}]
})
print("助手:", response2["messages"][-1].content)

5. 复杂工具组合的智能体

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
import math

# 定义多个专业工具
@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息。"""
    weather_data = {
        "北京": "晴天，温度 25°C，湿度 60%",
        "上海": "多云，温度 22°C，湿度 70%",
        "深圳": "小雨，温度 28°C，湿度 80%"
    }
    return weather_data.get(city, f"{city}的天气信息暂时无法获取")

@tool
def calculator(expression: str) -> str:
    """执行数学计算，支持基本运算和常用数学函数。"""
    try:
        result = eval(expression, {"__builtins__": {}}, {
            "sqrt": math.sqrt,
            "sin": math.sin,
            "cos": math.cos,
            "tan": math.tan,
            "pi": math.pi,
            "e": math.e,
            "log": math.log,
            "exp": math.exp
        })
        return str(result)
    except Exception as e:
        return f"计算错误: {str(e)}"

@tool
def get_time(timezone: str = "Asia/Shanghai") -> str:
    """获取当前时间。"""
    from datetime import datetime
    import pytz
    tz = pytz.timezone(timezone)
    return datetime.now(tz).strftime('%Y-%m-%d %H:%M:%S %Z')

# 创建智能体
agent = create_agent(
    model=ChatOpenAI(model="gpt-4o", temperature=0),
    tools=[get_weather, calculator, get_time],
    system_prompt="""你是一个多功能的智能助手，可以帮助用户：
    1. 查询天气信息
    2. 执行数学计算
    3. 查询当前时间
    
    请根据用户的问题，智能地选择和使用合适的工具。""",
)

# 复杂查询
response = agent.invoke({
    "messages": [{
        "role": "user",
        "content": """请帮我：
        1. 查询北京的天气
        2. 计算 sqrt(144) + 10*5 的结果
        3. 告诉我当前北京时间"""
    }]
})

print(response["messages"][-1].content)

---

完整实战案例

案例 1：智能客服助手

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

# 模拟产品数据库
products_db = {
    "iPhone 15": {"price": 5999, "stock": 100, "description": "最新款 iPhone"},
    "MacBook Pro": {"price": 12999, "stock": 50, "description": "专业级笔记本电脑"},
    "AirPods Pro": {"price": 1899, "stock": 200, "description": "降噪无线耳机"}
}

@tool
def search_product(product_name: str) -> str:
    """搜索产品信息。
    
    Args:
        product_name: 产品名称
    """
    product = products_db.get(product_name)
    if product:
        return f"产品：{product_name}\n价格：{product['price']}元\n库存：{product['stock']}件\n描述：{product['description']}"
    return f"未找到产品：{product_name}"

@tool
def check_order_status(order_id: str) -> str:
    """查询订单状态。
    
    Args:
        order_id: 订单编号
    """
    # 模拟订单查询
    return f"订单 {order_id} 的状态：已发货，预计明天送达"

@tool
def create_order(product_name: str, quantity: int) -> str:
    """创建订单。
    
    Args:
        product_name: 产品名称
        quantity: 购买数量
    """
    product = products_db.get(product_name)
    if not product:
        return f"产品 {product_name} 不存在"
    if product['stock'] < quantity:
        return f"库存不足，当前库存：{product['stock']}件"
    
    order_id = f"ORD{hash(product_name + str(quantity)) % 10000}"
    total_price = product['price'] * quantity
    return f"订单创建成功！\n订单号：{order_id}\n产品：{product_name}\n数量：{quantity}\n总价：{total_price}元"

# 创建客服智能体
agent = create_agent(
    model=ChatOpenAI(model="gpt-4o", temperature=0.3),
    tools=[search_product, check_order_status, create_order],
    system_prompt="""你是一个专业的电商客服助手。你的职责是：
    1. 帮助客户查询产品信息
    2. 查询订单状态
    3. 协助客户下单
    
    请用友好、专业的语气与客户交流。""",
)

# 模拟客服对话
queries = [
    "我想了解一下 iPhone 15 的信息",
    "帮我查一下订单 ORD1234 的状态",
    "我想买 2 个 AirPods Pro"
]

for query in queries:
    print(f"\n客户: {query}")
    response = agent.invoke({
        "messages": [{"role": "user", "content": query}]
    })
    print(f"客服: {response['messages'][-1].content}")

案例 2：数据分析助手

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
import pandas as pd
import numpy as np

# 模拟数据
data = pd.DataFrame({
    'date': pd.date_range('2024-01-01', periods=30),
    'sales': np.random.randint(1000, 5000, 30),
    'region': np.random.choice(['北京', '上海', '深圳'], 30)
})

@tool
def get_sales_summary() -> str:
    """获取销售数据摘要。"""
    summary = {
        '总销售额': data['sales'].sum(),
        '平均销售额': data['sales'].mean(),
        '最高销售额': data['sales'].max(),
        '最低销售额': data['sales'].min()
    }
    return str(summary)

@tool
def get_region_sales(region: str) -> str:
    """获取指定区域的销售数据。
    
    Args:
        region: 区域名称（北京、上海、深圳）
    """
    region_data = data[data['region'] == region]
    if region_data.empty:
        return f"未找到区域 {region} 的数据"
    
    summary = {
        '区域': region,
        '总销售额': region_data['sales'].sum(),
        '平均销售额': region_data['sales'].mean(),
        '数据点数量': len(region_data)
    }
    return str(summary)

@tool
def calculate_growth_rate(period: int = 7) -> str:
    """计算增长率。
    
    Args:
        period: 时间周期（天数），默认 7 天
    """
    recent = data.tail(period)['sales'].mean()
    previous = data.iloc[-period*2:-period]['sales'].mean()
    growth = ((recent - previous) / previous) * 100
    return f"最近 {period} 天相比前 {period} 天的增长率：{growth:.2f}%"

# 创建数据分析智能体
agent = create_agent(
    model=ChatOpenAI(model="gpt-4o", temperature=0),
    tools=[get_sales_summary, get_region_sales, calculate_growth_rate],
    system_prompt="""你是一个数据分析专家。你可以：
    1. 提供销售数据摘要
    2. 分析特定区域的销售情况
    3. 计算增长率等指标
    
    请用清晰、专业的方式呈现分析结果。""",
)

# 使用示例
response = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "请帮我分析一下销售情况，包括总体摘要、各区域对比和增长率"
    }]
})

print(response["messages"][-1].content)

案例 3：代码生成与执行助手

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
import subprocess
import sys

@tool
def execute_python_code(code: str) -> str:
    """执行 Python 代码并返回结果。
    
    Args:
        code: 要执行的 Python 代码
    """
    try:
        # 安全执行代码（实际应用中需要更严格的安全检查）
        result = eval(code)
        return f"执行结果：{result}"
    except Exception as e:
        return f"执行错误：{str(e)}"

@tool
def generate_code(description: str, language: str = "python") -> str:
    """根据描述生成代码。
    
    Args:
        description: 代码功能描述
        language: 编程语言，默认 python
    """
    # 这里应该调用代码生成模型或 API
    # 示例返回
    if language == "python":
        if "排序" in description:
            return "sorted_list = sorted(your_list)"
        elif "文件读取" in description:
            return "with open('file.txt', 'r') as f:\n    content = f.read()"
    return f"# {description}\n# 代码生成功能需要集成代码生成模型"

# 创建代码助手智能体
agent = create_agent(
    model=ChatOpenAI(model="gpt-4o", temperature=0.3),
    tools=[execute_python_code, generate_code],
    system_prompt="""你是一个编程助手，可以帮助用户：
    1. 生成代码
    2. 执行简单的 Python 代码
    3. 解答编程问题
    
    请确保生成的代码安全、高效、易读。""",
)

# 使用示例
response = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "帮我生成一个 Python 函数，用于计算斐波那契数列的第 n 项"
    }]
})

print(response["messages"][-1].content)

---

高级特性

1. 自定义系统提示词

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

custom_system_prompt = """你是一个专业的 AI 助手，具有以下特点：
1. 回答准确、专业
2. 语言简洁明了
3. 能够主动使用工具解决问题
4. 对不确定的问题会明确说明

请始终遵循以上原则。"""

agent = create_agent(
    model=ChatOpenAI(model="gpt-4o"),
    tools=[get_weather],
    system_prompt=custom_system_prompt,
)

2. 模型参数调优

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

# 高创造性任务
creative_agent = create_agent(
    model=ChatOpenAI(
        model="gpt-4o",
        temperature=0.9,  # 高温度，更创造性
        max_tokens=2000
    ),
    tools=[get_weather],
    system_prompt="你是一个富有创造力的助手",
)

# 精确性任务
precise_agent = create_agent(
    model=ChatOpenAI(
        model="gpt-4o",
        temperature=0,  # 低温度，更精确
        max_tokens=1000
    ),
    tools=[calculator],
    system_prompt="你是一个精确的计算助手",
)

3. 错误处理与重试

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
import time

@tool
def unreliable_api_call() -> str:
    """模拟不稳定的 API 调用。"""
    import random
    if random.random() < 0.5:
        raise Exception("API 调用失败")
    return "API 调用成功"

# 带重试的工具
@tool
def reliable_api_call(max_retries: int = 3) -> str:
    """带重试机制的 API 调用。"""
    for attempt in range(max_retries):
        try:
            return unreliable_api_call.invoke({})
        except Exception as e:
            if attempt == max_retries - 1:
                return f"重试 {max_retries} 次后仍然失败: {str(e)}"
            time.sleep(1)
    return "未知错误"

agent = create_agent(
    model=ChatOpenAI(model="gpt-4o"),
    tools=[reliable_api_call],
    system_prompt="你是一个可靠的助手",
)

4. 工具链式调用

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def get_user_location(user_id: str) -> str:
    """获取用户位置。"""
    locations = {
        "user1": "北京",
        "user2": "上海",
        "user3": "深圳"
    }
    return locations.get(user_id, "未知位置")

@tool
def get_weather(city: str) -> str:
    """获取天气信息。"""
    return f"{city}的天气：晴天，25°C"

# 智能体可以链式调用工具
agent = create_agent(
    model=ChatOpenAI(model="gpt-4o"),
    tools=[get_user_location, get_weather],
    system_prompt="你可以先获取用户位置，然后查询该位置的天气",
)

# 智能体会自动链式调用
response = agent.invoke({
    "messages": [{"role": "user", "content": "user1 所在城市的天气如何？"}]
})

print(response["messages"][-1].content)

---

最佳实践与注意事项

1. 工具设计原则

• 单一职责：每个工具应该只做一件事
• 清晰的文档字符串：工具的描述和参数说明要详细，帮助模型理解何时使用
• 错误处理：工具应该优雅地处理错误，返回有意义的错误信息
• 类型提示：使用类型提示帮助模型理解参数类型

@tool
def good_tool_example(city: str, unit: str = "celsius") -> str:
    """获取指定城市的天气信息。
    
    这是一个好的工具设计示例：
    - 有清晰的文档字符串
    - 参数有类型提示
    - 有默认值
    - 返回类型明确
    
    Args:
        city: 城市名称，例如：北京、上海
        unit: 温度单位，celsius（摄氏度）或 fahrenheit（华氏度），默认 celsius
    """
    # 实现逻辑
    return f"{city}的天气信息"

2. 系统提示词编写

• 明确角色：清楚定义智能体的角色和职责
• 行为规范：说明智能体应该如何行为
• 工具使用指导：告诉智能体何时使用哪些工具
• 输出格式：如果需要特定格式的输出，在提示词中说明

3. 性能优化

• 模型选择：根据任务复杂度选择合适的模型
• 温度设置：创造性任务用高温度，精确性任务用低温度
• 批量处理：对于多个相似请求，使用批量调用
• 缓存：对于重复的查询，考虑添加缓存机制

4. 安全考虑

• 输入验证：验证用户输入，防止注入攻击
• 工具权限：限制工具的执行权限
• 敏感信息：不要在工具中硬编码 API 密钥等敏感信息
• 代码执行：如果工具涉及代码执行，需要严格的安全检查

5. 调试技巧

• 启用 LangSmith：使用 LangSmith 追踪智能体的执行过程
• 日志记录：记录工具调用和模型响应
• 分步测试：先测试单个工具，再测试完整智能体
• 错误信息：提供详细的错误信息帮助调试

# 启用 LangSmith 追踪
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"

# 现在所有的调用都会被追踪
agent.invoke({"messages": [{"role": "user", "content": "测试"}]})