LangChain与LangGraph AI应用开发完全教程

从零到一掌握LangChain框架与LangGraph有状态图编排，构建生产级AI应用

关键词：LangChain, LangGraph, AI应用开发, LLM框架, Agent开发, RAG开发

引言：为什么选择LangChain与LangGraph
环境搭建与快速上手
LangChain核心模块详解
LCEL表达式语言深度解析
RAG Pipeline构建实战
对话记忆管理系统
Agent与工具调用机制
LangGraph有状态图编排
多Agent工作流设计
流式处理与回调机制
实战案例一：智能客服系统
实战案例二：研究助手系统
生产部署与监控
LangChain vs LangGraph对比分析
最佳实践与常见问题
总结与展望

1. 引言：为什么选择LangChain与LangGraph

1.1 AI应用开发的挑战

在大语言模型（LLM）蓬勃发展的今天，将模型能力转化为实际应用面临诸多挑战：

模型抽象层缺失：不同LLM提供商（OpenAI、Anthropic、Google等）的API各不相同，切换成本高
复杂流程编排困难：实际应用往往需要多步推理、条件分支、循环检查等复杂逻辑
状态管理复杂：对话上下文、中间结果、用户偏好等状态需要系统化管理
工具集成繁琐：与数据库、搜索引擎、外部API的集成需要大量胶水代码
可观测性不足：调试和监控LLM应用链路困难

1.2 LangChain的定位

LangChain是一个用于构建LLM应用的开源框架，它提供了：

统一的模型接口：封装不同LLM提供商，提供一致的调用方式
可组合的组件：Prompts、Chains、Retrievers、Memory等模块化组件
LCEL（LangChain Expression Language）：声明式链式调用语言
丰富的集成生态：600+集成组件覆盖向量数据库、工具、文档加载器等

1.3 LangGraph的定位

LangGraph是LangChain团队推出的有状态多Agent编排框架：

图结构编排：用有向图描述复杂的Agent工作流
持久化状态：内置状态管理与检查点机制
人机协作：原生支持Human-in-the-Loop模式
流式处理：支持节点级和token级的流式输出

1.4 两者的关系

LangChain提供基础组件和抽象，LangGraph在其之上提供复杂工作流编排能力。简单场景用LangChain Chains即可，复杂多步骤Agent场景则需要LangGraph。

2. 环境搭建与快速上手

2.1 安装依赖

# 创建虚拟环境
python -m venv langchain-env
source langchain-env/bin/activate  # Linux/Mac
# langchain-env\Scripts\activate   # Windows

# 安装核心包
pip install langchain langchain-core langchain-community langchain-openai
pip install langgraph
pip install chromadb  # 向量数据库
pip install tavily-python  # 搜索工具

2.2 配置API密钥

import os

# OpenAI API Key
os.environ["OPENAI_API_KEY"] = "sk-your-api-key"

# 如果使用其他提供商
os.environ["ANTHROPIC_API_KEY"] = "your-anthropic-key"
os.environ["TAVILY_API_KEY"] = "your-tavily-key"

安全提示：生产环境中请使用环境变量管理器或密钥管理服务，切勿将密钥硬编码在代码中。

2.3 第一个LangChain程序

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

# 初始化模型
llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0.7
)

# 直接调用
messages = [
    SystemMessage(content="你是一个专业的Python编程助手。"),
    HumanMessage(content="请解释Python中的装饰器是什么？")
]

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

2.4 第一个LangGraph程序

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

# 定义状态
class State(TypedDict):
    input: str
    output: str

# 定义节点函数
def process(state: State) -> dict:
    return {"output": f"处理结果: {state['input']}"}

# 构建图
graph = StateGraph(State)
graph.add_node("processor", process)
graph.add_edge(START, "processor")
graph.add_edge("processor", END)

# 编译并运行
app = graph.compile()
result = app.invoke({"input": "Hello LangGraph!"})
print(result["output"])

3. LangChain核心模块详解

3.1 Models（模型层）

LangChain的模型层提供了统一的接口来调用不同的LLM。

Chat Models（聊天模型）

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import (
    HumanMessage,
    SystemMessage,
    AIMessage
)

# OpenAI
openai_llm = ChatOpenAI(model="gpt-4o", temperature=0)

# Anthropic
claude_llm = ChatAnthropic(model="claude-sonnet-4-20250514", temperature=0)

# 使用统一接口调用
messages = [
    SystemMessage(content="你是一个有帮助的助手。"),
    HumanMessage(content="什么是机器学习？")
]

# 两者使用相同的消息格式
openai_response = openai_llm.invoke(messages)
claude_response = claude_llm.invoke(messages)

结构化输出

from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI

class MovieReview(BaseModel):
    """电影评价结构"""
    title: str = Field(description="电影名称")
    rating: float = Field(description="评分，0-10分", ge=0, le=10)
    summary: str = Field(description="一句话总结")
    pros: list[str] = Field(description="优点列表")
    cons: list[str] = Field(description="缺点列表")

llm = ChatOpenAI(model="gpt-4o-mini")
structured_llm = llm.with_structured_output(MovieReview)

review = structured_llm.invoke("请评价电影《盗梦空间》")
print(f"电影: {review.title}")
print(f"评分: {review.rating}/10")
print(f"总结: {review.summary}")

3.2 Prompts（提示模板）

PromptTemplate

from langchain_core.prompts import PromptTemplate

# 简单模板
template = PromptTemplate(
    input_variables=["product", "audience"],
    template="请为{product}写一段面向{audience}的营销文案，不超过100字。"
)

prompt = template.invoke({
    "product": "智能手表",
    "audience": "年轻白领"
})
print(prompt.text)

ChatPromptTemplate

from langchain_core.prompts import ChatPromptTemplate

# 多轮对话模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个{role}，请用{style}的语气回答问题。"),
    ("human", "{question}")
])

# 填充变量
messages = prompt.invoke({
    "role": "资深Python开发者",
    "style": "简洁专业",
    "question": "如何优化数据库查询性能？"
})

Few-Shot Prompting

from langchain_core.prompts import FewShotChatMessagePromptTemplate, ChatPromptTemplate

examples = [
    {"input": "今天天气真好", "output": "positive"},
    {"input": "服务太差了", "output": "negative"},
    {"input": "一般般吧", "output": "neutral"},
]

example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("ai", "{output}"),
])

few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_prompt=example_prompt,
    examples=examples,
)

final_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个情感分析专家，判断文本的情感倾向，只输出positive/negative/neutral。"),
    few_shot_prompt,
    ("human", "{input}"),
])

3.3 Chains（链）

链是LangChain中最基本的组合单元，将多个组件串联起来。

简单链（LCEL风格）

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

llm = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_template(
    "请将以下文本翻译成{language}：\n{text}"
)

# 使用LCEL构建链
chain = prompt | llm | StrOutputParser()

result = chain.invoke({
    "language": "英文",
    "text": "人工智能正在改变世界。"
})
print(result)

多步处理链

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

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

# 第一步：提取关键词
extract_prompt = ChatPromptTemplate.from_template(
    "从以下文本中提取3-5个关键词，用逗号分隔：\n{text}"
)

# 第二步：基于关键词生成摘要
summary_prompt = ChatPromptTemplate.from_template(
    "基于以下关键词，生成一段100字左右的摘要：\n{keywords}"
)

# 组合链
extract_chain = extract_prompt | llm | StrOutputParser()
summary_chain = summary_prompt | llm | StrOutputParser()

# 使用RunnablePassthrough传递数据
from langchain_core.runnables import RunnablePassthrough

full_chain = (
    {"keywords": extract_chain}
    | RunnablePassthrough.assign(
        summary=lambda x: summary_chain.invoke({"keywords": x["keywords"]})
    )
)

3.4 Output Parsers（输出解析器）

from langchain_core.output_parsers import (
    StrOutputParser,
    JsonOutputParser,
    CommaSeparatedListOutputParser
)
from langchain_core.prompts import ChatPromptTemplate

# 字符串解析
str_parser = StrOutputParser()

# JSON解析
json_parser = JsonOutputParser()

# 逗号分隔列表解析
list_parser = CommaSeparatedListOutputParser()

# 使用示例
prompt = ChatPromptTemplate.from_template(
    "列出5种常见的编程语言，用逗号分隔。\n{format_instructions}"
)

chain = prompt | ChatOpenAI(model="gpt-4o-mini") | list_parser
result = chain.invoke({
    "format_instructions": list_parser.get_format_instructions()
})
print(result)  # ['Python', 'JavaScript', 'Java', 'C++', 'Go']

3.5 Retrievers（检索器）

from langchain_community.document_loaders import TextLoader, DirectoryLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# 加载文档
loader = DirectoryLoader("./docs", glob="**/*.md", loader_cls=TextLoader)
documents = loader.load()

# 文本分割
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", "。", "！", "？", ".", " "]
)
chunks = splitter.split_documents(documents)

# 创建向量数据库
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(chunks, embeddings, persist_directory="./chroma_db")

# 创建检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 5}
)

# 检索
docs = retriever.invoke("LangChain的核心概念是什么？")
for doc in docs:
    print(f"来源: {doc.metadata.get('source', '未知')}")
    print(f"内容: {doc.page_content[:200]}...")
    print("---")

3.6 Memory（记忆）

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

# 创建消息历史存储
store = {}

def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

# 带记忆的链
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

chain = prompt | ChatOpenAI(model="gpt-4o-mini")

# 包装为带历史的链
with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history"
)

# 多轮对话
config = {"configurable": {"session_id": "user_123"}}
print(with_history.invoke({"input": "你好，我叫张三"}, config).content)
print(with_history.invoke({"input": "你还记得我叫什么名字吗？"}, config).content)

4. LCEL表达式语言深度解析

4.1 LCEL核心概念

LCEL（LangChain Expression Language）是LangChain的声明式链式调用语言。它使用管道操作符 | 将组件串联起来。

核心接口 - Runnable：

invoke：单次输入，单次输出
batch：批量输入，批量输出
stream：单次输入，流式输出
ainvoke/abatch/astream：异步版本

4.2 基础管道组合

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

# 经典的 Prompt | LLM | Parser 模式
prompt = ChatPromptTemplate.from_template("用一句话解释：{concept}")
llm = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

chain = prompt | llm | parser

# 调用方式
result = chain.invoke({"concept": "量子计算"})          # 单次
results = chain.batch([{"concept": "量子计算"}, {"concept": "区块链"}])  # 批量

# 流式输出
for chunk in chain.stream({"concept": "深度学习"}):
    print(chunk, end="", flush=True)

4.3 RunnablePassthrough与RunnableParallel

from langchain_core.runnables import RunnablePassthrough, RunnableParallel

# RunnablePassthrough - 透传输入
chain = RunnablePassthrough() | {"greeting": lambda x: f"Hello, {x['name']}!"}
result = chain.invoke({"name": "World"})
# {'greeting': 'Hello, World!'}

# RunnableParallel - 并行执行多个分支
paraphrase_prompt = ChatPromptTemplate.from_template("换一种说法：{text}")
summary_prompt = ChatPromptTemplate.from_template("总结以下内容：{text}")

parallel_chain = RunnableParallel(
    paraphrase=paraphrase_prompt | llm | parser,
    summary=summary_prompt | llm | parser
)

result = parallel_chain.invoke({"text": "LangChain是一个强大的LLM应用框架..."})
# result["paraphrase"] - 改写版本
# result["summary"] - 摘要版本

4.4 RunnableLambda - 自定义函数

from langchain_core.runnables import RunnableLambda

# 将普通函数包装为Runnable
def word_count(text: str) -> dict:
    return {"word_count": len(text.split()), "text": text}

word_counter = RunnableLambda(word_count)

# 在链中使用
chain = (
    ChatPromptTemplate.from_template("写一个关于{topic}的短文")
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
    | word_counter
)

result = chain.invoke({"topic": "人工智能"})
print(f"字数: {result['word_count']}")

4.5 条件分支与路由

from langchain_core.runnables import RunnableBranch, RunnableLambda

# 根据输入长度路由到不同的处理链
def route_by_length(input_data: dict) -> str:
    text = input_data["text"]
    if len(text) < 100:
        return "short"
    elif len(text) < 500:
        return "medium"
    else:
        return "long"

short_chain = ChatPromptTemplate.from_template("简短总结：{text}") | llm | parser
medium_chain = ChatPromptTemplate.from_template("详细总结：{text}") | llm | parser
long_chain = ChatPromptTemplate.from_template("提取要点并总结：{text}") | llm | parser

branch = RunnableBranch(
    (lambda x: route_by_length(x) == "short", short_chain),
    (lambda x: route_by_length(x) == "medium", medium_chain),
    long_chain,  # 默认分支
)

result = branch.invoke({"text": "这是一段很长的文本..."})

4.6 重试与回退机制

from langchain_core.runnables import RunnableRetry, RunnableWithFallbacks

# 重试机制
chain_with_retry = chain.with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True
)

# 回退机制 - 主模型失败时切换到备选模型
primary_llm = ChatOpenAI(model="gpt-4o", max_tokens=100)
fallback_llm = ChatOpenAI(model="gpt-4o-mini")

chain_with_fallback = primary_llm.with_fallbacks([fallback_llm])

5. RAG Pipeline构建实战

5.1 RAG架构概览

RAG（Retrieval-Augmented Generation）是将外部知识注入LLM的核心范式：

用户查询 → 查询改写 → 向量检索 → 文档重排序 → 上下文组装 → LLM生成 → 输出

5.2 文档加载与处理

from langchain_community.document_loaders import (
    PyPDFLoader,
    TextLoader,
    CSVLoader,
    UnstructuredMarkdownLoader,
    WebBaseLoader
)
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 加载PDF
pdf_loader = PyPDFLoader("knowledge_base/product_manual.pdf")
pdf_docs = pdf_loader.load()

# 加载网页
web_loader = WebBaseLoader("https://example.com/docs")
web_docs = web_loader.load()

# 合并所有文档
all_docs = pdf_docs + web_docs

# 智能文本分割
splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=800,
    chunk_overlap=150,
    separators=["\n\n", "\n", "。", "！", "？", "；", ".", " "]
)
chunks = splitter.split_documents(all_docs)

# 添加元数据
for i, chunk in enumerate(chunks):
    chunk.metadata["chunk_id"] = i
    chunk.metadata["char_count"] = len(chunk.page_content)

print(f"原始文档: {len(all_docs)} 个")
print(f"分割后: {len(chunks)} 个块")

5.3 向量存储与索引

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.retrievers import MultiQueryRetriever

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 创建向量数据库
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db",
    collection_name="knowledge_base"
)

# 基础检索器
base_retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 6}
)

# Multi-Query检索 - 用多个改写查询提高召回率
multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=base_retriever,
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
)

5.4 文档重排序

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder

# 使用Cross-Encoder重排序
cross_encoder = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-v2-m3")
reranker = CrossEncoderReranker(model=cross_encoder, top_n=3)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=reranker,
    base_retriever=multi_query_retriever
)

5.5 完整RAG Chain

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# RAG提示模板
rag_prompt = ChatPromptTemplate.from_template("""
你是一个专业的知识助手。请基于以下检索到的上下文回答用户问题。

规则：
1. 只基于提供的上下文回答，不要编造信息
2. 如果上下文中没有相关信息，请明确说明
3. 引用时请注明来源

上下文：
{context}

问题：{question}

回答：""")

# 格式化检索文档
def format_docs(docs):
    formatted = []
    for i, doc in enumerate(docs, 1):
        source = doc.metadata.get("source", "未知来源")
        formatted.append(f"[文档{i}] (来源: {source})\n{doc.page_content}")
    return "\n\n".join(formatted)

# 完整RAG链
rag_chain = (
    {
        "context": compression_retriever | format_docs,
        "question": RunnablePassthrough()
    }
    | rag_prompt
    | ChatOpenAI(model="gpt-4o")
    | StrOutputParser()
)

# 使用
answer = rag_chain.invoke("产品的退货政策是什么？")
print(answer)

5.6 带引用的RAG

from pydantic import BaseModel, Field

class Citation(BaseModel):
    source: str = Field(description="文档来源")
    page: int = Field(description="页码", default=0)
    quote: str = Field(description="引用原文")

class RAGResponse(BaseModel):
    answer: str = Field(description="回答内容")
    citations: list[Citation] = Field(description="引用列表")
    confidence: float = Field(description="置信度 0-1")

structured_rag = (
    {
        "context": retriever | format_docs,
        "question": RunnablePassthrough()
    }
    | rag_prompt
    | ChatOpenAI(model="gpt-4o").with_structured_output(RAGResponse)
)

response = structured_rag.invoke("产品的技术规格是什么？")
print(f"回答: {response.answer}")
for cite in response.citations:
    print(f"  引用: [{cite.source}] {cite.quote}")

6. 对话记忆管理系统

6.1 记忆类型概述

LangChain提供了多种记忆策略：

记忆类型	适用场景	特点
ConversationBufferMemory	短对话	保存完整历史
ConversationSummaryMemory	长对话	自动摘要压缩
ConversationBufferWindowMemory	滑动窗口	只保留最近N轮
VectorStoreRetrieverMemory	大量历史	基于相似度检索

6.2 基于LangChain的记忆实现

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.messages import HumanMessage, AIMessage

class ConversationManager:
    """对话记忆管理器"""

    def __init__(self, max_history: int = 20):
        self.histories: dict[str, ChatMessageHistory] = {}
        self.max_history = max_history

    def get_history(self, session_id: str) -> ChatMessageHistory:
        if session_id not in self.histories:
            self.histories[session_id] = ChatMessageHistory()
        return self.histories[session_id]

    def add_message(self, session_id: str, role: str, content: str):
        history = self.get_history(session_id)
        if role == "human":
            history.add_user_message(content)
        else:
            history.add_ai_message(content)

        # 超过最大长度时截断
        if len(history.messages) > self.max_history * 2:
            history.messages = history.messages[-(self.max_history * 2):]

    def get_context_window(self, session_id: str, window_size: int = 10) -> list:
        history = self.get_history(session_id)
        return history.messages[-(window_size * 2):]

    def clear(self, session_id: str):
        if session_id in self.histories:
            self.histories[session_id].clear()

6.3 带摘要的长期记忆

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

class SummaryMemory:
    """带摘要的长期记忆"""

    def __init__(self, llm=None):
        self.llm = llm or ChatOpenAI(model="gpt-4o-mini", temperature=0)
        self.summaries: dict[str, str] = {}
        self.recent_messages: dict[str, list] = {}

    def add_interaction(self, session_id: str, user_msg: str, ai_msg: str):
        if session_id not in self.recent_messages:
            self.recent_messages[session_id] = []

        self.recent_messages[session_id].append({
            "user": user_msg,
            "assistant": ai_msg
        })

        # 每5轮对话生成一次摘要
        if len(self.recent_messages[session_id]) >= 5:
            self._update_summary(session_id)

    def _update_summary(self, session_id: str):
        previous = self.summaries.get(session_id, "这是新的对话。")
        recent = self.recent_messages[session_id]

        conversation_text = "\n".join([
            f"用户: {m['user']}\n助手: {m['assistant']}"
            for m in recent
        ])

        summary_prompt = ChatPromptTemplate.from_template("""
请将以下对话历史与之前的摘要合并，生成新的简洁摘要（不超过200字）。

之前摘要：{previous_summary}

最近对话：
{recent_conversation}

新摘要：""")

        chain = summary_prompt | self.llm | StrOutputParser()
        new_summary = chain.invoke({
            "previous_summary": previous,
            "recent_conversation": conversation_text
        })

        self.summaries[session_id] = new_summary
        self.recent_messages[session_id] = []

    def get_context(self, session_id: str) -> str:
        summary = self.summaries.get(session_id, "")
        recent = self.recent_messages.get(session_id, [])

        context_parts = []
        if summary:
            context_parts.append(f"对话摘要：{summary}")
        for msg in recent[-3:]:
            context_parts.append(f"用户: {msg['user']}")
            context_parts.append(f"助手: {msg['assistant']}")

        return "\n".join(context_parts)

7. Agent与工具调用机制

7.1 工具定义

from langchain_core.tools import tool
from typing import Annotated

@tool
def search_web(query: Annotated[str, "搜索查询关键词"]) -> str:
    """在互联网上搜索最新信息。"""
    # 实际实现中调用搜索API
    return f"搜索结果: 关于'{query}'的最新信息..."

@tool
def calculator(expression: Annotated[str, "数学表达式"]) -> str:
    """计算数学表达式。"""
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"计算错误: {str(e)}"

@tool
def get_weather(city: Annotated[str, "城市名称"]) -> str:
    """获取指定城市的天气信息。"""
    # 实际实现中调用天气API
    weather_data = {
        "北京": "晴天，25°C",
        "上海": "多云，23°C",
        "广州": "阵雨，28°C"
    }
    return weather_data.get(city, f"暂无{city}的天气数据")

# 工具列表
tools = [search_web, calculator, get_weather]

7.2 ReAct Agent

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

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

# 使用LangGraph的预构建ReAct Agent
agent = create_react_agent(
    model=llm,
    tools=tools,
    prompt="你是一个智能助手，善于使用工具来回答用户的问题。"
)

# 调用Agent
result = agent.invoke({
    "messages": [("human", "北京今天天气怎么样？如果气温超过20度，推荐户外活动。")]
})

# 打印对话过程
for msg in result["messages"]:
    print(f"[{msg.type}] {msg.content}")

7.3 自定义Agent（LangGraph方式）

from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.prebuilt import ToolNode

# 定义工具节点
tool_node = ToolNode(tools)

# 定义Agent节点
def agent_node(state: MessagesState):
    response = llm.bind_tools(tools).invoke(state["messages"])
    return {"messages": [response]}

# 路由函数：判断是否需要调用工具
def should_continue(state: MessagesState):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

# 构建图
graph = StateGraph(MessagesState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)

graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")

app = graph.compile()

# 运行
result = app.invoke({
    "messages": [("human", "帮我计算 123 * 456 + 789")]
})

7.4 工具调用中的错误处理

from langchain_core.tools import tool
import tenacity

@tool
def robust_api_call(
    endpoint: Annotated[str, "API端点"],
    params: Annotated[str, "请求参数JSON"]
) -> str:
    """调用外部API，带重试机制。"""
    import json
    import time

    parsed_params = json.loads(params)
    max_retries = 3

    for attempt in range(max_retries):
        try:
            # 模拟API调用
            response = {"status": "success", "data": parsed_params}
            return json.dumps(response, ensure_ascii=False)
        except Exception as e:
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)
                continue
            return f"API调用失败: {str(e)}"

8. LangGraph有状态图编排

8.1 核心概念

LangGraph的核心概念包括：

State（状态）：图中流转的数据结构
Node（节点）：执行计算的函数单元
Edge（边）：节点之间的连接
Conditional Edge（条件边）：根据状态动态路由
Checkpoint（检查点）：状态持久化机制

8.2 状态定义

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    """Agent状态定义"""
    # 消息列表 - 使用add_messages进行追加合并
    messages: Annotated[list[BaseMessage], add_messages]

    # 当前步骤
    current_step: str

    # 检索到的文档
    documents: list[str]

    # 用户意图
    intent: str

    # 中间结果
    intermediate_results: dict

8.3 节点函数设计

def classify_intent(state: AgentState) -> dict:
    """意图分类节点"""
    last_message = state["messages"][-1].content

    classification_prompt = ChatPromptTemplate.from_template("""
根据用户消息判断意图类别：
- question: 提问咨询
- complaint: 投诉抱怨
- purchase: 购买相关
- support: 技术支持
- greeting: 问候寒暄

用户消息：{message}
只输出类别名称。""")

    chain = classification_prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()
    intent = chain.invoke({"message": last_message})

    return {"intent": intent.strip()}

def retrieve_knowledge(state: AgentState) -> dict:
    """知识检索节点"""
    query = state["messages"][-1].content
    docs = retriever.invoke(query)
    return {"documents": [doc.page_content for doc in docs]}

def generate_response(state: AgentState) -> dict:
    """生成回答节点"""
    context = "\n".join(state.get("documents", []))
    intent = state.get("intent", "general")

    prompt = ChatPromptTemplate.from_messages([
        ("system", f"""你是一个专业的客服助手。
当前用户意图: {intent}
知识库参考: {context}
请根据意图和知识库生成专业回答。"""),
        ("human", "{question}")
    ])

    chain = prompt | ChatOpenAI(model="gpt-4o") | StrOutputParser()
    response = chain.invoke({"question": state["messages"][-1].content})

    return {"messages": [("ai", response)]}

8.4 完整状态图构建

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

# 创建状态图
graph = StateGraph(AgentState)

# 添加节点
graph.add_node("classify", classify_intent)
graph.add_node("retrieve", retrieve_knowledge)
graph.add_node("respond", generate_response)
graph.add_node("direct_reply", lambda state: {
    "messages": [("ai", "您好！有什么可以帮助您的吗？")]
})

# 定义路由函数
def route_by_intent(state: AgentState) -> str:
    intent = state.get("intent", "")
    if intent == "greeting":
        return "direct_reply"
    else:
        return "retrieve"

# 添加边
graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_by_intent)
graph.add_edge("retrieve", "respond")
graph.add_edge("respond", END)
graph.add_edge("direct_reply", END)

# 编译（带检查点）
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# 运行（带线程ID）
config = {"configurable": {"thread_id": "session_001"}}
result = app.invoke(
    {"messages": [("human", "你们的退货政策是什么？")]},
    config=config
)

8.5 Human-in-the-Loop

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

# 在关键节点前设置中断
def human_approval(state: AgentState) -> dict:
    """等待人工审批"""
    return {}  # 不做任何操作，等待人工输入

graph = StateGraph(AgentState)
graph.add_node("classify", classify_intent)
graph.add_node("retrieve", retrieve_knowledge)
graph.add_node("human_review", human_approval)
graph.add_node("respond", generate_response)

graph.add_edge(START, "classify")
graph.add_edge("classify", "retrieve")
graph.add_edge("retrieve", "human_review")  # 中断点
graph.add_edge("human_review", "respond")
graph.add_edge("respond", END)

checkpointer = MemorySaver()
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["human_review"]  # 在此节点前中断
)

# 第一次运行 - 会在human_review前暂停
config = {"configurable": {"thread_id": "session_002"}}
result = app.invoke(
    {"messages": [("human", "我要退款")]},
    config=config
)

# 人工审查后继续
# 可以修改state后继续
app.update_state(config, {"documents": ["审查通过的知识文档"]})
result = app.invoke(None, config=config)  # 从中断点继续

9. 多Agent工作流设计

9.1 Supervisor模式

from langgraph.graph import StateGraph, START, END, MessagesState
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

# 专业化Agent
def research_agent(state: MessagesState):
    """研究Agent - 擅长信息收集"""
    llm = ChatOpenAI(model="gpt-4o").bind_tools([search_web])
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def analysis_agent(state: MessagesState):
    """分析Agent - 擅长数据分析"""
    llm = ChatOpenAI(model="gpt-4o")
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是数据分析专家，擅长从数据中发现模式和洞察。"),
        *state["messages"]
    ])
    response = (prompt | llm).invoke({})
    return {"messages": [response]}

def writing_agent(state: MessagesState):
    """写作Agent - 擅长内容创作"""
    llm = ChatOpenAI(model="gpt-4o")
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是专业的内容创作者，善于将复杂信息转化为清晰易懂的文字。"),
        *state["messages"]
    ])
    response = (prompt | llm).invoke({})
    return {"messages": [response]}

# Supervisor路由
def supervisor(state: MessagesState) -> str:
    """监督者 - 决定下一步由哪个Agent处理"""
    llm = ChatOpenAI(model="gpt-4o-mini")
    prompt = ChatPromptTemplate.from_messages([
        ("system", """你是工作流调度器。根据当前对话状态，决定下一步应该由哪个Agent处理：
- researcher: 需要收集更多信息时
- analyst: 需要分析已有信息时
- writer: 需要撰写最终报告时
- FINISH: 任务完成时

只输出Agent名称。"""),
        *state["messages"]
    ])
    result = (prompt | llm | StrOutputParser()).invoke({})
    return result.strip()

# 构建Supervisor图
workflow = StateGraph(MessagesState)

workflow.add_node("researcher", research_agent)
workflow.add_node("analyst", analysis_agent)
workflow.add_node("writer", writing_agent)
workflow.add_node("supervisor", lambda state: state)  # 空节点

workflow.add_edge(START, "supervisor")
workflow.add_conditional_edges(
    "supervisor",
    supervisor,
    {
        "researcher": "researcher",
        "analyst": "analyst",
        "writer": "writer",
        "FINISH": END
    }
)
workflow.add_edge("researcher", "supervisor")
workflow.add_edge("analyst", "supervisor")
workflow.add_edge("writer", "supervisor")

app = workflow.compile()

# 运行
result = app.invoke({
    "messages": [HumanMessage(content="请研究2024年AI行业发展趋势并撰写分析报告")]
})

9.2 协作Agent模式

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class CollaborationState(TypedDict):
    messages: Annotated[list, add_messages]
    code: str  # 代码
    review_comments: list[str]  # 审查意见
    iteration: int  # 迭代次数

def coder(state: CollaborationState):
    """编码Agent"""
    feedback = "\n".join(state.get("review_comments", []))
    prompt = f"""你是Python开发者。根据需求编写代码。
{f'请根据以下审查意见修改代码：{feedback}' if feedback else ''}
需求：{state['messages'][-1].content}"""

    llm = ChatOpenAI(model="gpt-4o")
    response = llm.invoke(prompt)
    return {
        "code": response.content,
        "iteration": state.get("iteration", 0) + 1
    }

def reviewer(state: CollaborationState):
    """审查Agent"""
    llm = ChatOpenAI(model="gpt-4o")
    prompt = f"""你是代码审查专家。审查以下代码，指出问题并给出改进建议。
如果代码质量足够好，输出 'APPROVED'。

代码：
{state['code']}"""

    response = llm.invoke(prompt)
    review = response.content

    if "APPROVED" in review:
        return {"review_comments": []}

    return {"review_comments": [review]}

def should_continue_review(state: CollaborationState):
    if not state.get("review_comments"):
        return END
    if state.get("iteration", 0) >= 3:
        return END
    return "coder"

# 构建协作图
graph = StateGraph(CollaborationState)
graph.add_node("coder", coder)
graph.add_node("reviewer", reviewer)

graph.add_edge(START, "coder")
graph.add_edge("coder", "reviewer")
graph.add_conditional_edges("reviewer", should_continue_review)

app = graph.compile()
result = app.invoke({
    "messages": [("human", "编写一个Python函数，实现快速排序算法")],
    "code": "",
    "review_comments": [],
    "iteration": 0
})

10. 流式处理与回调机制

10.1 Token级流式输出

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o", streaming=True)
prompt = ChatPromptTemplate.from_template("详细解释：{topic}")
chain = prompt | llm | StrOutputParser()

# Token级流式
async for token in chain.astream({"topic": "量子计算"}):
    print(token, end="", flush=True)

10.2 节点级流式输出（LangGraph）

from langgraph.graph import StateGraph, START, END, MessagesState

# 假设已有图定义app
# 节点级流式 - 每个节点完成时触发
async for event in app.astream_events(
    {"messages": [("human", "你好")]},
    version="v2"
):
    kind = event["event"]
    if kind == "on_chain_start":
        print(f"节点开始: {event['name']}")
    elif kind == "on_chain_end":
        print(f"节点完成: {event['name']}")
    elif kind == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="", flush=True)

10.3 自定义回调处理器

from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult

class LoggingCallbackHandler(BaseCallbackHandler):
    """自定义日志回调"""

    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"[LLM开始] 模型: {serialized.get('name', '未知')}")

    def on_llm_end(self, response: LLMResult, **kwargs):
        tokens = response.llm_output.get("token_usage", {}) if response.llm_output else {}
        print(f"[LLM结束] Token使用: {tokens}")

    def on_chain_start(self, serialized, inputs, **kwargs):
        print(f"[链开始] {serialized.get('name', '未知')}")

    def on_tool_start(self, serialized, input_str, **kwargs):
        print(f"[工具调用] {serialized.get('name', '未知')}: {input_str}")

    def on_tool_end(self, output, **kwargs):
        print(f"[工具返回] {output[:200]}")

    def on_retry(self, retry_state, **kwargs):
        print(f"[重试] 第{retry_state.attempt_number}次尝试")

# 使用回调
handler = LoggingCallbackHandler()
llm = ChatOpenAI(model="gpt-4o-mini", callbacks=[handler])
result = llm.invoke("Hello!")

10.4 生产级监控回调

import time
from langchain_core.callbacks import BaseCallbackHandler
from dataclasses import dataclass, field

@dataclass
class MetricsCollector:
    """指标收集器"""
    total_requests: int = 0
    total_tokens: int = 0
    total_latency_ms: float = 0
    errors: int = 0
    latencies: list = field(default_factory=list)

    def record(self, latency_ms: float, tokens: int = 0):
        self.total_requests += 1
        self.total_tokens += tokens
        self.total_latency_ms += latency_ms
        self.latencies.append(latency_ms)

    def get_p50_p99(self):
        sorted_lat = sorted(self.latencies)
        n = len(sorted_lat)
        if n == 0:
            return 0, 0
        p50 = sorted_lat[int(n * 0.5)]
        p99 = sorted_lat[int(n * 0.99)]
        return p50, p99

class MetricsCallbackHandler(BaseCallbackHandler):
    def __init__(self, collector: MetricsCollector):
        self.collector = collector
        self._start_times = {}

    def on_llm_start(self, serialized, prompts, **kwargs):
        self._start_times["llm"] = time.time()

    def on_llm_end(self, response: LLMResult, **kwargs):
        latency = (time.time() - self._start_times.get("llm", time.time())) * 1000
        tokens = 0
        if response.llm_output and "token_usage" in response.llm_output:
            tokens = response.llm_output["token_usage"].get("total_tokens", 0)
        self.collector.record(latency, tokens)

    def on_llm_error(self, error, **kwargs):
        self.collector.errors += 1

# 使用
metrics = MetricsCollector()
callback = MetricsCallbackHandler(metrics)
llm = ChatOpenAI(model="gpt-4o-mini", callbacks=[callback])

# ... 多次调用后
p50, p99 = metrics.get_p50_p99()
print(f"请求总数: {metrics.total_requests}")
print(f"P50延迟: {p50:.0f}ms, P99延迟: {p99:.0f}ms")
print(f"总Token: {metrics.total_tokens}")

11. 实战案例一：智能客服系统

11.1 系统架构

用户消息
    ↓
[意图识别] → 问候/投诉/咨询/购买
    ↓
[知识检索] → 向量搜索FAQ和文档
    ↓
[回复生成] → 结合上下文和知识库
    ↓
[质量检查] → 敏感词过滤、事实核查
    ↓
[人工转接] → 复杂问题转人工
    ↓
响应输出

11.2 完整实现

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.tools import tool
from pydantic import BaseModel
import json

# ========== 状态定义 ==========
class CustomerServiceState(TypedDict):
    messages: Annotated[list, add_messages]
    intent: str           # 意图分类
    sentiment: str        # 情感倾向
    customer_info: dict   # 客户信息
    knowledge_docs: list  # 检索到的知识
    escalate: bool        # 是否转人工
    resolution: str       # 解决方案

# ========== 意图识别 ==========
class Intent(BaseModel):
    category: Literal["greeting", "complaint", "inquiry", "purchase", "return", "technical", "other"]
    confidence: float

def classify_intent(state: CustomerServiceState) -> dict:
    """意图识别节点"""
    llm = ChatOpenAI(model="gpt-4o-mini")
    structured_llm = llm.with_structured_output(Intent)

    prompt = ChatPromptTemplate.from_messages([
        ("system", """你是意图分类专家。将客户消息分类为以下类别：
- greeting: 问候寒暄
- complaint: 投诉抱怨
- inquiry: 咨询询问
- purchase: 购买相关
- return: 退换货
- technical: 技术问题
- other: 其他"""),
        ("human", "{message}")
    ])

    chain = prompt | structured_llm
    result = chain.invoke({"message": state["messages"][-1].content})

    return {"intent": result.category}

# ========== 情感分析 ==========
def analyze_sentiment(state: CustomerServiceState) -> dict:
    """情感分析节点"""
    llm = ChatOpenAI(model="gpt-4o-mini")
    prompt = ChatPromptTemplate.from_template("""
分析以下客户消息的情感倾向，只输出：positive/neutral/negative/angry

客户消息：{message}
情感：""")

    chain = prompt | llm | StrOutputParser()
    sentiment = chain.invoke({"message": state["messages"][-1].content})

    return {"sentiment": sentiment.strip().lower()}

# ========== 知识检索 ==========
def retrieve_knowledge(state: CustomerServiceState) -> dict:
    """知识库检索节点"""
    query = state["messages"][-1].content
    intent = state.get("intent", "")

    # 根据意图增强查询
    enhanced_query = f"{intent}: {query}"
    docs = retriever.invoke(enhanced_query)

    return {"knowledge_docs": [doc.page_content for doc in docs]}

# ========== 路由逻辑 ==========
def route_after_analysis(state: CustomerServiceState) -> str:
    """分析后的路由"""
    sentiment = state.get("sentiment", "")
    intent = state.get("intent", "")

    # 高愤怒情绪直接转人工
    if sentiment == "angry":
        return "escalate"

    # 投诉类优先转人工
    if intent == "complaint" and sentiment == "negative":
        return "escalate"

    return "generate"

# ========== 生成回复 ==========
def generate_response(state: CustomerServiceState) -> dict:
    """回复生成节点"""
    knowledge = "\n".join(state.get("knowledge_docs", []))
    intent = state.get("intent", "general")
    sentiment = state.get("sentiment", "neutral")

    # 根据情感调整语气
    tone_map = {
        "positive": "热情友好",
        "neutral": "专业礼貌",
        "negative": "耐心安抚",
        "angry": "诚恳道歉、积极解决"
    }
    tone = tone_map.get(sentiment, "专业礼貌")

    system_prompt = f"""你是一位专业的客服代表。

当前客户意图: {intent}
情感状态: {sentiment}
语气要求: {tone}

参考知识库：
{knowledge if knowledge else "无特定参考，请根据通用知识回答。"}

回复规则：
1. 保持{tone}的语气
2. 基于知识库内容回答，不要编造
3. 如果无法解决，建议转接人工客服
4. 回复控制在200字以内"""

    llm = ChatOpenAI(model="gpt-4o")
    messages = [SystemMessage(content=system_prompt)] + state["messages"]
    response = llm.invoke(messages)

    return {"messages": [response]}

# ========== 人工转接 ==========
def escalate_to_human(state: CustomerServiceState) -> dict:
    """转接人工客服"""
    return {
        "messages": [AIMessage(content="非常抱歉给您带来不便，我正在为您转接人工客服，请稍候...")],
        "escalate": True
    }

# ========== 质量检查 ==========
def quality_check(state: CustomerServiceState) -> dict:
    """质量检查节点 - 检查回复是否合规"""
    last_response = state["messages"][-1].content

    # 简单的敏感词检查
    sensitive_words = ["保证", "一定", "绝对", "承诺"]
    has_sensitive = any(word in last_response for word in sensitive_words)

    if has_sensitive:
        # 重新生成更谨慎的回复
        llm = ChatOpenAI(model="gpt-4o-mini")
        rewrite_prompt = ChatPromptTemplate.from_template("""
请将以下客服回复改写得更谨慎，避免过度承诺：
原始回复：{response}
改写：""")
        chain = rewrite_prompt | llm | StrOutputParser()
        new_response = chain.invoke({"response": last_response})
        return {"messages": [AIMessage(content=new_response)]}

    return {}

# ========== 构建完整图 ==========
graph = StateGraph(CustomerServiceState)

# 添加节点
graph.add_node("classify_intent", classify_intent)
graph.add_node("analyze_sentiment", analyze_sentiment)
graph.add_node("retrieve", retrieve_knowledge)
graph.add_node("generate", generate_response)
graph.add_node("quality_check", quality_check)
graph.add_node("escalate", escalate_to_human)

# 定义边
graph.add_edge(START, "classify_intent")
graph.add_edge("classify_intent", "analyze_sentiment")
graph.add_conditional_edges("analyze_sentiment", route_after_analysis)
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", "quality_check")
graph.add_edge("quality_check", END)
graph.add_edge("escalate", END)

# 编译
checkpointer = MemorySaver()
customer_service_app = graph.compile(checkpointer=checkpointer)

# ========== 使用示例 ==========
config = {"configurable": {"thread_id": "customer_001"}}

# 模拟对话
conversations = [
    "你好，请问你们的发货时间是多久？",
    "我三天前下的单，到现在还没收到！太慢了！",
    "我想退货，买的衣服尺码不合适",
]

for msg in conversations:
    print(f"\n客户: {msg}")
    result = customer_service_app.invoke(
        {"messages": [HumanMessage(content=msg)]},
        config=config
    )
    print(f"客服: {result['messages'][-1].content}")
    print(f"意图: {result.get('intent')} | 情感: {result.get('sentiment')}")

11.3 会话管理与持久化

from langgraph.checkpoint.postgres import PostgresSaver

# 使用PostgreSQL持久化（生产环境）
# 需要安装: pip install langgraph-checkpoint-postgres

# 连接字符串
DB_URI = "postgresql://user:password@localhost:5432/langgraph_db"

# 创建持久化检查点
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # 首次运行创建表

    app = graph.compile(checkpointer=checkpointer)

    # 会话状态自动持久化
    result = app.invoke(
        {"messages": [HumanMessage(content="你好")]},
        config={"configurable": {"thread_id": "session_xyz"}}
    )

    # 后续可以从断点恢复
    # app.get_state(config)  # 获取当前状态

12. 实战案例二：研究助手系统

12.1 系统设计

研究助手系统采用多Agent协作架构：

Research Agent：搜索和收集信息
Analysis Agent：分析和总结信息
Citation Agent：管理引用和来源
Report Agent：撰写最终报告
Supervisor：协调各Agent工作

12.2 实现代码

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, SystemMessage
from pydantic import BaseModel
import json

# ========== 工具定义 ==========
@tool
def web_search(query: str) -> str:
    """搜索互联网获取最新信息。"""
    # 实际使用Tavily等搜索API
    import urllib.request
    # 简化示例
    return f"搜索'{query}'的结果：相关最新信息..."

@tool
def save_notes(note_title: str, content: str) -> str:
    """保存研究笔记。"""
    # 实际实现中写入文件或数据库
    return f"笔记'{note_title}'已保存。"

@tool
def get_saved_notes(topic: str) -> str:
    """获取之前保存的研究笔记。"""
    return f"关于'{topic}'的笔记：暂无保存的笔记。"

research_tools = [web_search, save_notes, get_saved_notes]

# ========== 状态定义 ==========
class ResearchState(TypedDict):
    messages: Annotated[list, add_messages]
    topic: str                  # 研究主题
    research_findings: list     # 研究发现
    analysis: str               # 分析结果
    citations: list             # 引用列表
    report: str                 # 最终报告
    current_phase: str          # 当前阶段

# ========== Research Agent ==========
def research_agent(state: ResearchState):
    """研究Agent - 搜索和收集信息"""
    llm = ChatOpenAI(model="gpt-4o").bind_tools(research_tools)

    system = SystemMessage(content=f"""你是研究助手，负责收集关于"{state['topic']}"的信息。
请执行以下步骤：
1. 搜索3-5个不同角度的信息
2. 保存重要的研究发现
3. 确保信息来源多样可靠""")

    messages = [system] + state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}

# ========== Analysis Agent ==========
def analysis_agent(state: ResearchState):
    """分析Agent - 分析和总结"""
    llm = ChatOpenAI(model="gpt-4o")

    findings_text = "\n".join(state.get("research_findings", []))
    prompt = ChatPromptTemplate.from_messages([
        ("system", f"""你是数据分析专家。请对以下研究发现进行深入分析：

研究主题：{state['topic']}
研究发现：
{findings_text if findings_text else "请基于对话历史进行分析"}

分析要求：
1. 识别关键趋势和模式
2. 提炼核心洞察
3. 指出数据中的矛盾或空白
4. 给出结论性判断"""),
        *state["messages"]
    ])

    response = (prompt | llm).invoke({})
    return {
        "analysis": response.content,
        "messages": [response]
    }

# ========== Report Agent ==========
def report_agent(state: ResearchState):
    """报告Agent - 撰写最终报告"""
    llm = ChatOpenAI(model="gpt-4o")

    prompt = ChatPromptTemplate.from_messages([
        ("system", f"""你是专业研究报告撰写者。请根据以下内容撰写一份结构化的研究报告：

研究主题：{state['topic']}
分析结果：{state.get('analysis', '暂无')}
引用来源：{json.dumps(state.get('citations', []), ensure_ascii=False)}

报告结构：
1. 摘要（200字）
2. 背景介绍
3. 主要发现
4. 深入分析
5. 结论与建议
6. 参考资料

请使用Markdown格式。"""),
        *state["messages"]
    ])

    response = (prompt | llm).invoke({})
    return {
        "report": response.content,
        "messages": [response]
    }

# ========== 路由逻辑 ==========
def supervisor_route(state: ResearchState) -> str:
    """监督者路由"""
    phase = state.get("current_phase", "research")

    phase_map = {
        "research": "researcher",
        "analysis": "analyzer",
        "report": "reporter",
        "complete": END
    }

    # 自动推进阶段
    if phase == "research" and state.get("research_findings"):
        return "analyzer"
    elif phase == "analysis" and state.get("analysis"):
        return "reporter"
    elif phase == "report" and state.get("report"):
        return END

    return phase_map.get(phase, "researcher")

# ========== 构建研究助手图 ==========
graph = StateGraph(ResearchState)

graph.add_node("researcher", research_agent)
graph.add_node("analyzer", analysis_agent)
graph.add_node("reporter", report_agent)

graph.add_edge(START, "researcher")
graph.add_conditional_edges("researcher", lambda s: "analyzer" if s.get("research_findings") else END)
graph.add_conditional_edges("analyzer", lambda s: "reporter" if s.get("analysis") else END)
graph.add_edge("reporter", END)

research_app = graph.compile()

# ========== 使用示例 ==========
result = research_app.invoke({
    "messages": [HumanMessage(content="请研究2024年大语言模型的发展趋势")],
    "topic": "2024年大语言模型发展趋势",
    "research_findings": [],
    "analysis": "",
    "citations": [],
    "report": "",
    "current_phase": "research"
})

print("=== 研究报告 ===")
print(result.get("report", "报告生成中..."))

13. 生产部署与监控

13.1 FastAPI部署

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langserve import add_routes
import uvicorn

app = FastAPI(title="AI应用API", version="1.0.0")

# 请求模型
class ChatRequest(BaseModel):
    message: str
    session_id: str = "default"
    user_id: str = "anonymous"

class ChatResponse(BaseModel):
    reply: str
    intent: str = ""
    metadata: dict = {}

# 使用LangServe添加LangChain路由
add_routes(
    app,
    rag_chain,
    path="/rag",
    playground_type="chat"
)

# 自定义聊天端点
@app.post("/chat", response_model=ChatResponse)
async def chat_endpoint(request: ChatRequest):
    try:
        config = {"configurable": {"thread_id": request.session_id}}
        result = customer_service_app.invoke(
            {"messages": [("human", request.message)]},
            config=config
        )

        return ChatResponse(
            reply=result["messages"][-1].content,
            intent=result.get("intent", ""),
            metadata={
                "sentiment": result.get("sentiment", ""),
                "escalated": result.get("escalate", False)
            }
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# 流式端点
from fastapi.responses import StreamingResponse
from langchain_core.messages import HumanMessage

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    async def generate():
        config = {"configurable": {"thread_id": request.session_id}}
        async for event in customer_service_app.astream_events(
            {"messages": [HumanMessage(content=request.message)]},
            config=config,
            version="v2"
        ):
            if event["event"] == "on_chat_model_stream":
                token = event["data"]["chunk"].content
                if token:
                    yield f"data: {token}\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

# 健康检查
@app.get("/health")
async def health_check():
    return {"status": "healthy", "version": "1.0.0"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

13.2 LangSmith监控集成

import os

# 配置LangSmith
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-ai-app"

# 配置后，所有LangChain调用自动追踪
# 在LangSmith控制台可以看到：
# - 每次调用的完整链路
# - 每个节点的输入输出
# - Token使用量和延迟
# - 错误和异常

13.3 性能优化策略

from langchain_core.runnables import RunnableLambda
from functools import lru_cache
import hashlib

# 1. 缓存策略
class SemanticCache:
    """语义缓存 - 相似查询复用结果"""

    def __init__(self, embeddings, threshold=0.95):
        self.embeddings = embeddings
        self.threshold = threshold
        self.cache = {}  # embedding -> response

    def get(self, query: str):
        query_emb = self.embeddings.embed_query(query)
        for cached_emb, response in self.cache.items():
            similarity = self._cosine_similarity(query_emb, cached_emb)
            if similarity > self.threshold:
                return response
        return None

    def set(self, query: str, response: str):
        query_emb = tuple(self.embeddings.embed_query(query))
        self.cache[query_emb] = response

    def _cosine_similarity(self, a, b):
        import numpy as np
        a, b = np.array(a), np.array(b)
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# 2. 并行执行
parallel_chain = RunnableParallel(
    context=retriever | format_docs,
    history=lambda x: get_session_history(x["session_id"]),
    metadata=lambda x: get_user_metadata(x["user_id"])
)

# 3. 批量处理
async def process_batch(messages: list[str]):
    results = await chain.abatch(
        [{"question": msg} for msg in messages],
        return_exceptions=True
    )
    return results

13.4 Docker部署

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用
COPY . .

# 环境变量
ENV PYTHONPATH=/app
ENV LANGCHAIN_TRACING_V2=true

# 健康检查
HEALTHCHECK --interval=30s --timeout=10s \
    CMD curl -f http://localhost:8000/health || exit 1

EXPOSE 8000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: '3.8'
services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - LANGCHAIN_API_KEY=${LANGCHAIN_API_KEY}
    depends_on:
      - redis
      - postgres

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  postgres:
    image: postgres:15-alpine
    environment:
      POSTGRES_DB: langgraph
      POSTGRES_USER: admin
      POSTGRES_PASSWORD: ${DB_PASSWORD}
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

14. LangChain vs LangGraph对比分析

14.1 设计理念对比

维度	LangChain	LangGraph
核心抽象	Chain（链）	Graph（图）
数据流	线性管道	有向图，支持循环
状态管理	外部Memory组件	内置持久化状态
适用复杂度	简单到中等	中等到复杂
人机协作	需手动实现	原生支持
流式粒度	Token级	节点级+Token级

14.2 何时选择LangChain

适合LangChain的场景：

简单的Prompt → LLM → Output流程
单轮RAG问答
快速原型验证
不需要复杂状态管理

# LangChain适合的简单场景
chain = prompt | llm | parser
result = chain.invoke({"input": "你好"})

14.3 何时选择LangGraph

适合LangGraph的场景：

多步骤推理和决策
需要循环或条件分支
多Agent协作
需要持久化状态和断点恢复
Human-in-the-Loop场景

# LangGraph适合的复杂场景
graph = StateGraph(State)
graph.add_node("step1", ...)
graph.add_node("step2", ...)
graph.add_conditional_edges("step1", router)
# ... 复杂编排

14.4 混合使用

在实际项目中，两者可以混合使用：

# 用LangChain构建可复用的组件
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | parser
)

# 用LangGraph编排复杂工作流
graph = StateGraph(AppState)
graph.add_node("rag", lambda state: {"answer": rag_chain.invoke(state["query"])})
graph.add_node("agent", agent_node)
# ...

15. 最佳实践与常见问题

15.1 Prompt工程最佳实践

角色定义清晰：System Prompt中明确Agent的角色和能力边界
输出格式约束：使用结构化输出（with_structured_output）确保格式一致
Few-Shot示例：提供输入输出示例提高一致性
链式思考：对复杂推理任务引导模型逐步思考

15.2 常见问题

Q1: 如何处理Rate Limit？

from langchain_core.rate_limiters import InMemoryRateLimiter

rate_limiter = InMemoryRateLimiter(
    requests_per_second=10,
    check_every_n_seconds=0.1,
    max_bucket_size=20
)

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

Q2: 如何节省Token成本？

# 1. 使用小模型处理简单任务
classify_llm = ChatOpenAI(model="gpt-4o-mini")  # 分类用小模型
generate_llm = ChatOpenAI(model="gpt-4o")        # 生成用大模型

# 2. 控制上下文长度
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})  # 减少检索数量

# 3. 使用摘要记忆替代完整历史
from langchain.memory import ConversationSummaryMemory

Q3: 如何测试Agent？

import pytest

@pytest.mark.asyncio
async def test_customer_service():
    """测试客服Agent"""
    result = customer_service_app.invoke(
        {"messages": [HumanMessage(content="你好")]},
        config={"configurable": {"thread_id": "test_001"}}
    )

    assert result["intent"] == "greeting"
    assert result["sentiment"] in ["positive", "neutral"]
    assert len(result["messages"][-1].content) > 0

@pytest.mark.asyncio
async def test_escalation():
    """测试人工转接"""
    result = customer_service_app.invoke(
        {"messages": [HumanMessage(content="你们的服务太差了！我要投诉！")]},
        config={"configurable": {"thread_id": "test_002"}}
    )

    assert result.get("escalate") == True

16. 总结与展望

16.1 核心要点回顾

LangChain 提供了LLM应用开发的基础组件和抽象，适合快速构建标准化流程
LangGraph 在其之上提供了复杂工作流编排能力，适合多Agent和有状态场景
LCEL 是连接两者的粘合剂，声明式管道让代码更简洁
RAG 是将外部知识注入LLM的核心范式，需要关注检索质量和上下文管理
生产部署 需要考虑监控、缓存、限流、持久化等工程化问题

16.2 学习路径建议

入门：掌握LangChain基础组件和LCEL
进阶：学习RAG和Agent开发
高级：掌握LangGraph有状态图编排
专家：多Agent系统设计和生产级部署

16.3 推荐资源

版权声明：本教程为原创内容，仅供学习参考。代码示例基于LangChain和LangGraph的公开API编写。

LangChain与LangGraph AI应用开发完全教程

教程简介

LangChain与LangGraph AI应用开发完全教程

目录

1. 引言：为什么选择LangChain与LangGraph

1.1 AI应用开发的挑战

1.2 LangChain的定位

1.3 LangGraph的定位

1.4 两者的关系

2. 环境搭建与快速上手

2.1 安装依赖

2.2 配置API密钥

2.3 第一个LangChain程序

2.4 第一个LangGraph程序

3. LangChain核心模块详解

3.1 Models（模型层）

Chat Models（聊天模型）

结构化输出

3.2 Prompts（提示模板）

PromptTemplate

ChatPromptTemplate

Few-Shot Prompting

3.3 Chains（链）

简单链（LCEL风格）

多步处理链

3.4 Output Parsers（输出解析器）

3.5 Retrievers（检索器）

3.6 Memory（记忆）

4. LCEL表达式语言深度解析

4.1 LCEL核心概念

4.2 基础管道组合

4.3 RunnablePassthrough与RunnableParallel

4.4 RunnableLambda - 自定义函数

4.5 条件分支与路由

4.6 重试与回退机制

5. RAG Pipeline构建实战

5.1 RAG架构概览

5.2 文档加载与处理

5.3 向量存储与索引

5.4 文档重排序

5.5 完整RAG Chain

5.6 带引用的RAG

6. 对话记忆管理系统

6.1 记忆类型概述

6.2 基于LangChain的记忆实现

6.3 带摘要的长期记忆

7. Agent与工具调用机制

7.1 工具定义

7.2 ReAct Agent

7.3 自定义Agent（LangGraph方式）

7.4 工具调用中的错误处理

8. LangGraph有状态图编排

8.1 核心概念

8.2 状态定义

8.3 节点函数设计

8.4 完整状态图构建

8.5 Human-in-the-Loop

9. 多Agent工作流设计

9.1 Supervisor模式

9.2 协作Agent模式

10. 流式处理与回调机制

10.1 Token级流式输出

10.2 节点级流式输出（LangGraph）

10.3 自定义回调处理器

10.4 生产级监控回调

11. 实战案例一：智能客服系统

11.1 系统架构

11.2 完整实现

11.3 会话管理与持久化

12. 实战案例二：研究助手系统

12.1 系统设计

12.2 实现代码

13. 生产部署与监控

13.1 FastAPI部署

13.2 LangSmith监控集成

13.3 性能优化策略

13.4 Docker部署

14. LangChain vs LangGraph对比分析

14.1 设计理念对比

14.2 何时选择LangChain