Python LangChain 完整学习手册 v3

目录

第一部分：LangChain 基础

1. Hello LangChain - 第一个 LLM 调用
2. 提示词模板使用
3. 消息类型与对话管理
4. 自定义工具
5. Simple Agent
6. Agent Loop - Agent 执行循环

第二部分：进阶功能

1. Memory Basics - 内存管理基础
2. Context Management - 上下文管理
3. Checkpointing - 检查点持久化
4. Middleware Basics - 中间件基础
5. Structured Output - 结构化输出
6. Validation & Retry - 验证和重试

第三部分：RAG 系统

1. RAG Basics - RAG 基础
2. RAG Advanced - RAG 进阶

第四部分：LangGraph 工作流

1. LangGraph 基础
2. Multi-Agent - 多 Agent 协作
3. Conditional Routing - 条件路由

第五部分：多模态处理

1. Image Input - 图像输入
2. File Handling - 文件处理
3. Mixed Modality - 混合模态

第六部分：生产部署

1. Tools & Agents 进阶
2. LangSmith 集成
3. Error Handling - 错误处理

第一部分：LangChain 基础

1. Hello LangChain - 第一个 LLM 调用

1.1 使用 Groq API 调用 Kimi-K2 模型

from langchain_groq import ChatGroq
import os
from dotenv import load_dotenv

loaded = load_dotenv()
GROQ_KEY = os.getenv("GROQ_API_KEY")

model = ChatGroq(
    model='moonshotai/kimi-k2-instruct',
    api_key=GROQ_KEY,
    temperature=0.7,
)
responses = model.invoke("请用一句话介绍一下你自己")
print(responses.content)

1.2 使用阿里云百炼调用千问模型

from langchain.chat_models import init_chat_model
import os
from dotenv import load_dotenv

loaded = load_dotenv(dotenv_path="../.env")
QIANWEN_KEY = os.getenv("QIANWEN_API_KEY")

model = init_chat_model(
    model='openai:qwen3.5-flash',
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key=QIANWEN_KEY,
    temperature=0.7,
)
responses = model.invoke("用一句话介绍一下你自己")
print(responses.content)

1.3 多模型配置（千问、智谱、百度）

from langchain.chat_models import init_chat_model
import os
from dotenv import load_dotenv

load_dotenv()

# 千问配置
qwen_config = {
    "api_key": os.getenv("QIANWEN_API_KEY"),
    "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
    "model_name": "qwen3.5-flash",
}

# 智谱配置
zhipu_config = {
    "api_key": os.getenv("ZHIPU_API_KEY"),
    "base_url": "https://open.bigmodel.cn/api/paas/v4",
    "model_name": "glm-4.7",
}

# 百度配置
baidu_config = {
    "api_key": os.getenv("BAIDU_API_KEY"),
    "base_url": "https://qianfan.baidubce.com/v2",
    "model_name": "deepseek-v3.1-250821",
}

# 配置使用千问模型
model_config = qwen_config

chat_model = init_chat_model(
    model=model_config["model_name"],
    api_key=model_config["api_key"],
    model_provider="openai",
    base_url=model_config["base_url"],
    temperature=0.7,
)

1.4 关键知识点

2. 提示词模板使用

2.1 PromptTemplate - 简单文本模板

from langchain_core.prompts import PromptTemplate

temp1 = PromptTemplate.from_template("你是一名成语解释专家，请帮我简单解释一下这个成语{word}")
prompt = temp1.format(word="画蛇添足")
response = chat_model.invoke(prompt)
print(response.content)

2.2 ChatPromptTemplate - 聊天消息模板

from langchain_core.prompts import ChatPromptTemplate

# 使用元组格式（推荐）
template = ChatPromptTemplate.from_messages([
    ("system", "你是一个{role},请一句话介绍一下我的问题"),
    ("user", "{question}")
])

messages = template.format_messages(
    role="Python 导师",
    question="什么是装饰器？"
)
response = chat_model.invoke(messages)
print(response.content)

2.3 LCEL 链式调用

from langchain_core.prompts import ChatPromptTemplate

# 创建组件
template = ChatPromptTemplate.from_messages([
    ("system", "你是{role}"),
    ("user", "{input}")
])

# 使用 | 创建链
chain = template | chat_model

# 直接调用链
response = chain.invoke({
    "role": "Python 导师",
    "input": "什么是装饰器？,一句话解释"
})
print(response.content)

2.4 关键知识点

3. 消息类型与对话管理

3.1 多轮对话示例

conversation = [
    {"role": "system", "content": "你是一个成语解释器"}
]

# 第一轮对话
conversation.append({"role": "user", "content": "什么是高高兴兴？请一句话解释"})
r1 = chat_model.invoke(conversation)
print(r1.content)

# 关键：保存 AI 回复
conversation.append({"role": "assistant", "content": r1.content})

# 第二轮
conversation.append({"role": "user", "content": "我刚刚问的是哪一个成语？"})
r1 = chat_model.invoke(conversation)
print(r1.content)

3.2 关键知识点

• 字典格式：{"role": "system/user/assistant", "content": "..."} 推荐
• 保存 AI 回复：conversation.append({"role": "assistant", "content": r.content})
• 对话历史：每次调用必须传递完整消息列表

4. 自定义工具

4.1 使用 @tool 装饰器

from langchain.tools import tool

@tool
def get_weather(city: str) -> str:
    """
    获取指定城市的天气信息
    参数:
        city (str): 要查询天气的城市名称 例如 "北京"

    返回:
        str: 包含天气信息的字符串
    """
    weather_data = {
        "北京": "晴天，温度 15°C，空气质量良好",
        "上海": "多云，温度 18°C，有轻微雾霾",
        "深圳": "阴天，温度 22°C，可能有小雨",
        "成都": "小雨，温度 12°C，湿度较高",
    }
    if city not in weather_data:
        return f"城市 {city} 的天气信息未找到"
    return weather_data.get(city, f"抱歉，暂时没有{city}的天气数据")


@tool
def calculator(operation: str, a: float, b: float) -> float:
    """
    执行基本的数学计算

    参数:
        operation: 运算类型，支持 "add"(加), "subtract"(减), "multiply"(乘), "divide"(除)
        a: 第一个数字
        b: 第二个数字

    返回:
        计算结果字符串
    """
    operations = {
        "add": lambda x, y: x + y,
        "subtract": lambda x, y: x - y,
        "multiply": lambda x, y: x * y,
        "divide": lambda x, y: x / y if y != 0 else "错误：除数不能为零",
    }
    result = operations[operation](a, b)
    return float(result)

4.2 绑定工具到模型

model_with_tools = chat_model.bind_tools([get_weather])

# AI 可以决定是否调用工具
response = model_with_tools.invoke("北京天气怎么样？")

if response.tool_calls:
    print("AI 想调用工具：", response.tool_calls)
else:
    print("AI 直接回答：", response.content)

4.3 关键知识点

5. Simple Agent

5.1 创建 Agent

from langchain.agents import create_agent

agent = create_agent(
    model=chat_model,
    tools=[get_weather, calculator],
)

sendMsg = {
    "messages": [{"role": "user", "content": "10 + 5 等于多少"}]
}

response = agent.invoke(sendMsg)

5.2 遍历消息流程

for msg in response["messages"]:
    msg_type = msg.__class__.__name__
    if msg_type == "HumanMessage":
        print(f"\n[HumanMessage] 用户输入")
        print(f"  内容: {msg.content}")
    elif msg_type == "AIMessage":
        if hasattr(msg, "tool_calls") and msg.tool_calls:
            print(f"\n[AIMessage] AI 决定调用工具")
            print(f"  工具: {msg.tool_calls[0]['name']}")
            print(f"  参数: {msg.tool_calls[0]['args']}")
        else:
            print(f"\n[AIMessage] AI 的最终回答")
            print(f"  内容: {msg.content}")
    elif msg_type == "ToolMessage":
        print(f"\n[ToolMessage] 工具执行结果")
        print(f"  工具: {msg.name}")
        print(f"  结果: {msg.content}")

5.3 关键知识点

• 执行循环：HumanMessage → AIMessage(tool_calls) → ToolMessage → AIMessage(最终回答)
• 获取最终回答：response['messages'][-1].content
• 工具数量：建议 2-5 个工具效果最佳

6. Agent Loop - Agent 执行循环

6.1 流式输出

# invoke() - 完整结果，等待完成后一次性获取
response = agent.invoke(sendMsg)

# stream() - 流式输出，实时获取中间步骤
for chunk in agent.stream(sendMsg):
    print("-" * 50)
    print(chunk)
    print("-" * 50)

6.2 执行流程

用户输入
    ↓
AIMessage(tool_calls)  ← AI 决定调用工具
    ↓
ToolMessage  ← 工具执行结果
    ↓
AIMessage(最终回答)  ← AI 根据结果回答

第二部分：进阶功能

7. Memory Basics - 内存管理基础

7.1 使用 InMemorySaver

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model=chat_model,
    tools=[],
    checkpointer=InMemorySaver()  # 添加内存
)

# 调用时指定 thread_id
config = {"configurable": {"thread_id": "user_1"}}

# 第一轮
inputMsg = {"messages": [{"role": "user", "content": "我叫张三"}]}
agent.invoke(input=inputMsg, config=config)

# 第二轮 - 记得第一轮！
inputMsg2 = {"messages": [{"role": "user", "content": "我刚刚说我叫什么？"}]}
response2 = agent.invoke(input=inputMsg2, config=config)
print(f"Agent: {response2['messages'][-1].content}")

7.2 关键知识点

8. Context Management - 上下文管理

8.1 使用 SummarizationMiddleware

from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model=chat_model,
    tools=[],
    middleware=[
        SummarizationMiddleware(
            model=chat_model,
            max_messages=2,  # 保留最近消息数
            trigger_tokens=30  # 触发阈值
        )
    ]
)

8.2 手动修剪消息

from langchain_core.messages import trim_messages

trimmed = trim_messages(
    messages,
    max_tokens=1000,
    strategy="last",
    start_on="human"
)

9. Checkpointing - 检查点持久化

9.1 使用 SqliteSaver

from langgraph.checkpoint.sqlite import SqliteSaver

# 创建持久化检查点
with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    agent = create_agent(
        model=chat_model,
        tools=[get_weather, calculator],
        checkpointer=checkpointer
    )
    
    config = {"configurable": {"thread_id": "user_session_1"}}
    
    # 所有对话历史自动持久化
    response = agent.invoke(inputMsg, config=config)

9.2 关键知识点

10. Middleware Basics - 中间件基础

10.1 自定义中间件

from langchain.agents.middleware import AgentMiddleware

class LoggingMiddleware(AgentMiddleware):
    def before_model(self, state):
        print("  [日志] 模型调用前...")
        return None  # 继续流程
    
    def after_model(self, state):
        print("  [日志] 模型调用后...")
        return None

agent = create_agent(
    model=chat_model,
    tools=[],
    middleware=[LoggingMiddleware()]
)

10.2 内置中间件

11. Structured Output - 结构化输出

11.1 使用 Pydantic 模型

from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model

class Person(BaseModel):
    """用户信息"""
    name: str = Field(description="用户姓名")
    age: int = Field(description="用户年龄")
    email: str = Field(description="用户邮箱")

model = init_chat_model("openai:qwen3.5-flash", ...)
structured_model = model.with_structured_output(Person)

# 返回 Pydantic 对象
person = structured_model.invoke("张三，25岁，邮箱 zhangsan@example.com")
print(person.name)  # 张三
print(person.age)   # 25

11.2 嵌套模型

class InvoiceItem(BaseModel):
    product: str
    quantity: int
    price: float

class Invoice(BaseModel):
    invoice_id: str
    items: List[InvoiceItem]
    total: float

12. Validation & Retry - 验证和重试

12.1 with_retry() - 自动重试

llm_with_retry = model.with_retry(
    retry_if_exception_type=(ConnectionError, TimeoutError),
    wait_exponential_jitter=True,
    stop_after_attempt=7
)

try:
    response = llm_with_retry.invoke("你好")
except Exception as e:
    print("请求失败：", str(e))

12.2 with_fallbacks() - 降级方案

test_model = init_chat_model("groq:llama-3.3-70b-versatile")
real_model = init_chat_model("openai:qwen3.5-flash", ...)

llm_with_fallbacks = test_model.with_fallbacks([real_model])

response = llm_with_fallbacks.invoke("一句话介绍 Python")

12.3 Pydantic 验证

from pydantic import BaseModel, field_validator

class User(BaseModel):
    name: str
    age: int
    email: str

    @field_validator('name')
    def name_min_length(cls, v):
        if len(v) < 2:
            raise ValueError('String should have at least 2 characters')
        return v

    @field_validator('age')
    def age_max_limit(cls, v):
        if v > 150:
            raise ValueError('Input should be less than or equal to 150')
        return v

    @field_validator('email')
    def email_must_contain_at(cls, v):
        if '@' not in v:
            raise ValueError('邮箱格式无效，必须包含 @ 符号')
        return v

第三部分：RAG 系统

13. RAG Basics - RAG 基础

13.1 完整 RAG 流程

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_pinecone import PineconeVectorStore

# 1. 文档加载
loader = TextLoader("document.txt")
documents = loader.load()

# 2. 文本分割
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
chunks = splitter.split_documents(documents)

# 3. 向量嵌入
embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)

# 4. 向量存储
vector_store = PineconeVectorStore.from_documents(
    documents=chunks,
    embedding=embeddings,
    index_name="my-index"
)

# 5. 相似度检索
results = vector_store.similarity_search("查询内容", k=3)

13.2 RAG 工作流

文档加载 → 文本分割 → 向量嵌入 → 向量存储
    ↓
查询 → 相似度检索 → 提供给 LLM → 生成答案

14. RAG Advanced - RAG 进阶

14.1 向量检索（语义搜索）

# 创建检索器
retriever = vector_store.as_retriever(search_kwargs={"k": 3})
results = retriever.invoke("光明")

优点：

• 能懂 "言外之意"
• 认得出同义词
• 支持跨语言

缺点：

• 对精准关键词不敏感
• 专有名词匹配差

14.2 BM25 检索（关键词搜索）

from langchain_community.retrievers import BM25Retriever
import jieba

# 中文分词
def chinese_tokenizer(text):
    return list(jieba.cut(text))

bm25_retriever = BM25Retriever.from_documents(
    documents,
    preprocess_func=chinese_tokenizer
)
bm25_retriever.k = 3
results = bm25_retriever.invoke("失败")

优点：

• 精准匹配
• 速度快
• 不依赖嵌入模型

缺点：

• 不懂语义
• 同义词不认

14.3 混合检索（EnsembleRetriever）

from langchain_classic.retrievers import EnsembleRetriever

ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]  # BM25 40%, 向量 60%
)

results = ensemble_retriever.invoke("接受失败")

14.4 权重配置建议

第四部分：LangGraph 工作流

15. LangGraph 基础

15.1 什么是 LangGraph？

LangGraph 是一个用于构建状态化、多步骤 AI 应用的框架：

• 节点（Nodes）：图中的处理单元
• 边（Edges）：连接节点的路径
• 状态（State）：在节点之间传递的数据结构

15.2 Hello World Graph

from typing import TypedDict
from langgraph.graph import StateGraph

class AgentState(TypedDict):
    message: str

def greeting_node(state: AgentState) -> AgentState:
    state['message'] = "Hello" + state["message"]
    return state

graph = StateGraph(AgentState)
graph.add_node("greeter", greeting_node)
graph.set_entry_point("greeter")
graph.set_finish_point("greeter")
app = graph.compile()

result = app.invoke({"message": " langgraph"})
print(result['message'])  # Hello langgraph

15.3 多节点顺序执行

class AgentState(TypedDict):
    name: str
    age: str
    final: str

def first_node(state: AgentState) -> AgentState:
    state["final"] = f"你好 {state['name']}"
    return state

def second_node(state: AgentState) -> AgentState:
    state["final"] = state["final"] + f"你已经 {state['age']}了"
    return state

graph = StateGraph(AgentState)
graph.add_node("first_node", first_node)
graph.add_node("second_node", second_node)

graph.set_entry_point("first_node")
graph.add_edge("first_node", "second_node")
graph.set_finish_point("second_node")
app = graph.compile()

result = app.invoke({"name": "zhangsan", "age": "18"})
# {'name': 'zhangsan', 'age': '18', 'final': '你好 zhangsan你已经 18了'}

15.4 条件分支

from typing import Literal

def route_by_score(state: AgentState) -> Literal["excellent", "good", "improve", "reject"]:
    score = state["score"]
    if score >= 90:
        return "excellent"
    elif score >= 70:
        return "good"
    elif score >= 50:
        return "improve"
    else:
        return "reject"

graph.add_conditional_edges(
    "evaluate",
    route_by_score,
    {
        "excellent": "excellent",
        "good": "good",
        "improve": "improve",
        "reject": "reject"
    }
)

16. Multi-Agent - 多 Agent 协作

16.1 三种协作模式

监督者模式（Supervisor Pattern）

                    ┌──────────────┐
                    │  Supervisor  │
                    │   (协调者)    │
                    └──────┬───────┘
           ┌───────────────┼───────────────┐
           ▼               ▼               ▼
    ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
    │  Agent A    │ │  Agent B    │ │  Agent C    │
    │  (研究员)    │ │  (编辑)     │ │  (审核员)   │
    └─────────────┘ └─────────────┘ └─────────────┘

协作链模式（Collaborative Pattern）

    ┌─────────────┐     ┌─────────────┐
    │  Agent A    │────▶│  Agent B    │
    │  (写初稿)    │     │  (审核修改)  │
    └─────────────┘     └──────┬──────┘
                               │
                               ▼
                        ┌─────────────┐
                        │  Agent C    │
                        │  (最终确认)  │
                        └─────────────┘

动态分发模式

def classifier(state: SupportState) -> dict:
    """分类器：识别问题类型"""
    # 分析用户问题，返回分类
    if "退款" in state["query"]:
        return {"category": "billing"}
    elif "bug" in state["query"]:
        return {"category": "technical"}
    else:
        return {"category": "general"}

def route_to_specialist(state) -> Literal["billing", "technical", "general"]:
    return state["category"]

16.2 完整监督者模式示例

class TeamState(TypedDict):
    task: str
    messages: Annotated[list, add_messages]
    research_result: str
    draft: str
    final_content: str
    next_agent: str

def supervisor(state: TeamState) -> dict:
    if not state.get("research_result"):
        next_agent = "researcher"
    elif not state.get("draft"):
        next_agent = "writer"
    elif not state.get("final_content"):
        next_agent = "editor"
    else:
        next_agent = "complete"
    return {"next_agent": next_agent}

# 各 Agent 完成后回到 supervisor
graph.add_edge("researcher", "supervisor")
graph.add_edge("writer", "supervisor")
graph.add_edge("editor", "supervisor")

17. Conditional Routing - 条件路由

17.1 内容评分路由系统

def score_based_routing():
    """
    根据评分决定处理流程
    - 优秀 (>= 90)：发送表扬信
    - 良好 (>= 70)：正常通过
    - 需改进 (>= 50)：提供建议
    - 不合格 (< 50)：需要重新提交
    """
    def evaluate(state: AgentState) -> dict:
        # 评估内容并打分
        response = model.invoke(messages)
        score = int(response.content.strip())
        return {"score": score}

    def route_by_score(state: AgentState) -> Literal["excellent", "good", "improve", "reject"]:
        score = state["score"]
        if score >= 90:
            return "excellent"
        elif score >= 70:
            return "good"
        elif score >= 50:
            return "improve"
        else:
            return "reject"

17.2 重试机制

def retry_mechanism():
    """
    实现带重试的工作流
    - 任务可能随机失败
    - 最多重试 3 次
    - 超过重试次数后使用备用方案
    """
    def should_retry(state: RetryState) -> Literal["retry", "fallback", "success"]:
        if state["success"]:
            return "success"
        if state["retry_count"] < state["max_retries"]:
            return "retry"
        return "fallback"

    graph.add_conditional_edges(
        "execute",
        should_retry,
        {
            "retry": "execute",      # 重试：回到执行节点
            "fallback": "fallback",  # 备用方案
            "success": "success"     # 成功
        }
    )

第五部分：多模态处理

18. Image Input - 图像输入

18.1 图像编码

import base64
from pathlib import Path

def encode_image_to_base64(image_path: str) -> str:
    """将本地图像编码为 base64"""
    with open(image_path, "rb") as image_file:
        return base64.standard_b64encode(image_file.read()).decode("utf-8")

def get_mime_type(image_path: str) -> str:
    """根据文件扩展名获取 MIME 类型"""
    ext = Path(image_path).suffix.lower()
    mime_types = {
        ".jpg": "image/jpeg",
        ".jpeg": "image/jpeg",
        ".png": "image/png",
        ".gif": "image/gif",
        ".webp": "image/webp"
    }
    return mime_types.get(ext, "image/jpeg")

18.2 创建图像消息

from langchain_core.messages import HumanMessage

def create_image_message(text: str, image_path: str) -> HumanMessage:
    image_base64 = encode_image_to_base64(image_path)
    mime_type = get_mime_type(image_path)

    content = [
        {"type": "text", "text": text},
        {
            "type": "image_url",
            "image_url": {"url": f"data:{mime_type};base64,{image_base64}"}
        }
    ]
    return HumanMessage(content=content)

# 使用
message = create_image_message(
    text="请详细描述这张图片中的内容。用中文回复。",
    image_path="haibao.jpg"
)
response = model.invoke([message])

18.3 图像问答

questions = [
    "提取图片中的采购单号",
    "提取图片中的供货商",
]

messages = [create_image_message("我会问你关于这张图片的一些问题。", image_path)]

for question in questions:
    messages.append(HumanMessage(content=question))
    response = model.invoke(messages)
    messages.append(response)
    print(f"回答: {response.content}")

19. File Handling - 文件处理

19.1 文档问答

# 读取文件
with open("./ai.txt", "r", encoding="utf-8") as f:
    content = f.read()

questions = [
    "ai的技术是什么？",
    "AI的应用场景有哪些？",
]

for question in questions:
    messages = [
        SystemMessage(content=f"""你是一个文档问答助手。根据以下文档内容回答问题。
如果文档中没有相关信息，请说明"文档中未提及此信息"。
用中文简洁回答。

文档内容：
{content}"""),
        HumanMessage(content=question)
    ]
    response = model.invoke(messages)

19.2 多文件合并分析

def multi_file_analysis(txt_path: str, csv_path: str, json_path: str):
    documents = []

    # 文本文件
    with open(txt_path, "r", encoding="utf-8") as f:
        documents.append(Document(
            page_content=f.read()[:1000],
            metadata={"source": "python_guide.txt", "type": "tutorial"}
        ))

    # CSV 文件
    with open(csv_path, "r", encoding="utf-8") as f:
        documents.append(Document(
            page_content=f.read(),
            metadata={"source": "employees.csv", "type": "data"}
        ))

    # 合并内容
    combined_content = "\n\n---\n\n".join([
        f"【{doc.metadata['source']}】\n{doc.page_content}"
        for doc in documents
    ])

20. Mixed Modality - 混合模态

20.1 文本 + 图像混合输入

content = [
    {"type": "text", "text": """以下是我们公司的销售数据：

**2024年第一季度销售报告**
- 1月: 150万
- 2月: 180万  
- 3月: 220万

请结合图表分析：
1. 数据趋势如何？
2. 与图表显示的趋势是否一致？
3. 你有什么建议？"""},
    create_image_content("chart.png")
]

message = HumanMessage(content=content)
response = model.invoke([message])

20.2 多图像对比分析

content = [
    {"type": "text", "text": "请对比这两张图片，说明它们的相同点和不同点。"},
    create_image_content("image1.jpg"),
    create_image_content("image2.jpg")
]

message = HumanMessage(content=content)
response = model.invoke([message])

20.3 LangGraph 混合模态处理

class MultimodalState(TypedDict):
    text_input: str
    image_paths: List[str]
    analysis_result: Optional[str]
    summary: Optional[str]

def analyze_content(state: MultimodalState) -> MultimodalState:
    content = [{"type": "text", "text": state["text_input"]}]
    for img_path in state["image_paths"]:
        if os.path.exists(img_path):
            content.append(create_image_content(img_path))
    
    message = HumanMessage(content=content)
    response = model.invoke([message])
    state["analysis_result"] = response.content
    return state

第六部分：生产部署

21. Tools & Agents 进阶

21.1 带验证的工具

from pydantic import BaseModel, Field, field_validator

class EmailInput(BaseModel):
    """邮件发送参数"""
    to: str = Field(description="收件人邮箱")
    subject: str = Field(description="邮件主题")
    body: str = Field(description="邮件正文")

    @field_validator('to')
    @classmethod
    def validate_email(cls, v):
        if '@' not in v:
            raise ValueError('无效的邮箱地址')
        return v

@tool(args_schema=EmailInput)
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件（带参数验证）"""
    return f"邮件已发送至 {to}"

21.2 Agent 回调监控

from langchain_core.callbacks import BaseCallbackHandler

class AgentMonitor(BaseCallbackHandler):
    """监控 Agent 执行"""

    def on_tool_start(self, tool_name, tool_input, **kwargs):
        print(f"🔧 工具开始: {tool_name}")
        print(f"   输入: {tool_input}")

    def on_tool_end(self, output, **kwargs):
        print(f"✅ 工具完成: {output[:100]}...")

response = agent.invoke(
    {"messages": [...]},
    config={"callbacks": [AgentMonitor()]}
)

22. LangSmith 集成

22.1 环境变量配置

# .env 文件
LANGSMITH_API_KEY=your_api_key_here
LANGSMITH_TRACING=true
LANGSMITH_PROJECT=my-project-name

22.2 自动追踪

import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "my-project"

# LangChain 会自动发送追踪数据
model = init_chat_model("openai:gpt-4o-mini")
response = model.invoke("Hello!")

22.3 手动标记

from langsmith import traceable

@traceable(name="my_function", tags=["production"])
def my_custom_function(input_data):
    return result

22.4 添加元数据

from langchain_core.runnables import RunnableConfig

config = RunnableConfig(
    metadata={
        "user_id": "user_123",
        "session_id": "sess_456"
    },
    tags=["production", "v2"]
)

response = model.invoke("Hello!", config=config)

23. Error Handling - 错误处理

23.1 常见错误类型

23.2 完整错误处理示例

# 1. 重试机制
llm_with_retry = model.with_retry(
    retry_if_exception_type=(ConnectionError, TimeoutError),
    wait_exponential_jitter=True,
    stop_after_attempt=3
)

# 2. 降级方案
llm_with_fallbacks = primary_model.with_fallbacks([backup_model])

# 3. 输出验证
class User(BaseModel):
    name: str
    age: int
    
    @field_validator('age')
    def validate_age(cls, v):
        if v < 0 or v > 150:
            raise ValueError('年龄必须在 0-150 之间')
        return v

# 4. 综合使用
try:
    response = llm_with_retry.invoke(prompt)
    user = User.model_validate(response)
except ValidationError as e:
    print("验证失败:", e)
except Exception as e:
    print("其他错误:", e)

总结

本手册整合了 LangChain 学习项目的所有核心内容：

学习路径

1. 基础阶段：Module 01-06

   • LLM 调用、提示词、消息、工具、Agent

2. 进阶阶段：Module 07-12

   • 内存、上下文、检查点、中间件、结构化输出、验证重试

3. RAG 阶段：Module 13-14

   • RAG 基础、混合检索

4. LangGraph 阶段：Module 15-17

   • 状态图、多 Agent、条件路由

5. 多模态阶段：Module 18-20

   • 图像输入、文件处理、混合模态

6. 生产阶段：Module 21-23

   • 工具进阶、LangSmith、错误处理

核心要点

• LangChain 1.0 新 API：init_chat_model、create_agent
• LangGraph：状态化工作流的核心
• RAG：向量检索 + BM25 混合检索
• 多模态：文本、图像、文件统一处理
• 生产部署：重试、降级、监控、错误处理

所有示例都来自实际运行的代码，可直接复制使用！