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

从零基础到生产部署，一站式掌握 LangChain 与 LangGraph 的核心能力

LangChain 概述与安装
核心概念总览
LLM 与 ChatModel 使用
Prompt Template 与 Few-shot
Output Parser 输出解析
Chain 链式调用
Agent 智能体
Tool 工具系统开发
Memory 记忆管理
LangGraph 核心概念与 StateGraph
LangGraph 节点与边的定义
多 Agent 协作模式
RAG 集成实战
流式输出处理
生产部署最佳实践
实战项目：智能客服系统

1. LangChain 概述与安装

1.1 什么是 LangChain

LangChain 是一个用于构建大语言模型（LLM）应用的开源框架。它提供了一套标准化的抽象层，让开发者可以轻松地将 LLM 与外部数据源、工具和工作流进行集成。其核心价值在于：

模块化设计：每个组件可独立使用，也可灵活组合
生态丰富：支持数百种模型、向量数据库和工具的集成
生产就绪：内置链路追踪、错误处理和可观测性支持

1.2 安装与环境配置

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

# 安装核心包
pip install langchain langchain-core langchain-community

# 安装特定模型提供者（按需选择）
pip install langchain-openai        # OpenAI
pip install langchain-anthropic     # Anthropic
pip install langchain-ollama        # 本地 Ollama

# 安装 LangGraph
pip install langgraph

# 安装常用工具包
pip install langchainhub chromadb   # 向量数据库
pip install tavily-python           # 搜索工具

1.3 环境变量设置

import os

# 推荐使用 .env 文件管理密钥
# pip install python-dotenv
from dotenv import load_dotenv
load_dotenv()

# 或直接设置
os.environ["OPENAI_API_KEY"] = "your-api-key-here"
os.environ["TAVILY_API_KEY"] = "your-tavily-key"

⚠️ 安全提示：永远不要将 API Key 硬编码在代码中，使用环境变量或密钥管理服务。

2. 核心概念总览

LangChain 的架构围绕四大核心概念构建：

┌─────────────────────────────────────────────────┐
│                  LangChain 生态                   │
├──────────┬──────────┬──────────┬─────────────────┤
│ Model I/O│  Chains  │  Agents  │     Memory      │
│ 模型输入输出 │  链式调用  │  智能体   │     记忆管理    │
├──────────┴──────────┴──────────┴─────────────────┤
│              Tools & Retrieval                    │
│              工具系统与检索                         │
├─────────────────────────────────────────────────┤
│              LangGraph（编排层）                   │
└─────────────────────────────────────────────────┘

Model I/O：标准化的模型输入（Prompt）→ 模型调用（LLM）→ 输出解析（Parser）流程
Chains：将多个组件串联成可复用的工作流
Agents：让 LLM 自主决策使用哪些工具、按什么顺序执行
Memory：为无状态的 LLM 添加上下文记忆能力

3. LLM 与 ChatModel 使用

LangChain 提供两种模型接口：LLM（文本补全）和 ChatModel（对话补全）。

3.1 ChatModel 基础使用

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

# 初始化模型
llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0.7,      # 控制随机性，0-2
    max_tokens=1024,      # 最大输出 token 数
    timeout=30,           # 请求超时（秒）
)

# 方式一：直接调用
response = llm.invoke("什么是量子计算？")
print(response.content)

# 方式二：使用消息列表（推荐）
messages = [
    SystemMessage(content="你是一位物理学教授，用通俗语言解释概念。"),
    HumanMessage(content="什么是量子纠缠？"),
]
response = llm.invoke(messages)
print(response.content)

3.2 使用本地模型（Ollama）

from langchain_ollama import ChatOllama

llm = ChatOllama(
    model="qwen2.5:7b",   # Ollama 已下载的模型
    temperature=0.7,
)

response = llm.invoke("用Python写一个快速排序")
print(response.content)

3.3 模型参数调优

# 使用 with_structured_output 约束输出格式
from pydantic import BaseModel, Field

class MovieReview(BaseModel):
    title: str = Field(description="电影名称")
    rating: float = Field(description="评分，1-10")
    summary: str = Field(description="一句话总结")
    recommend: bool = Field(description="是否推荐")

structured_llm = llm.with_structured_output(MovieReview)
review = structured_llm.invoke("评价一下《盗梦空间》这部电影")
print(review.title)    # 盗梦空间
print(review.rating)   # 9.0

4. Prompt Template 与 Few-shot

4.1 基础 Prompt Template

from langchain_core.prompts import ChatPromptTemplate

# 简单模板
prompt = ChatPromptTemplate.from_template(
    "请将以下文本翻译成{language}：\n\n{text}"
)
formatted = prompt.invoke({"language": "法语", "text": "你好世界"})
print(formatted)

4.2 多角色对话模板

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位{role}，擅长用{style}的方式回答问题。"),
    ("human", "{question}"),
])

chain = prompt | llm
result = chain.invoke({
    "role": "Python 教师",
    "style": "幽默风趣",
    "question": "装饰器是什么？"
})
print(result.content)

4.3 Few-shot 提示

from langchain_core.prompts import FewShotChatMessagePromptTemplate

examples = [
    {"input": "今天天气真好", "output": "正面"},
    {"input": "这个产品太差了", "output": "负面"},
    {"input": "一般般吧", "output": "中性"},
    {"input": "服务态度非常棒！", "output": "正面"},
]

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", "你是情感分析专家，判断文本情感：正面、负面或中性。"),
    few_shot_prompt,
    ("human", "{input}"),
])

chain = final_prompt | llm
result = chain.invoke({"input": "这家餐厅的菜品很一般，但服务不错"})
print(result.content)  # 中性

5. Output Parser 输出解析

5.1 StrOutputParser（最常用）

from langchain_core.output_parsers import StrOutputParser

chain = prompt | llm | StrOutputParser()
result = chain.invoke({"input": "你好"})
# result 直接是字符串，而非 AIMessage

5.2 JsonOutputParser

from langchain_core.output_parsers import JsonOutputParser

parser = JsonOutputParser(pydantic_object=MovieReview)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是影评专家。{format_instructions}"),
    ("human", "评价电影：{movie}"),
])

chain = prompt.partial(format_instructions=parser.get_format_instructions()) | llm | parser
result = chain.invoke({"movie": "星际穿越"})
print(type(result))  # <class 'dict'>

5.3 自定义输出解析器

from langchain_core.output_parsers import BaseOutputParser

class CommaSeparatedParser(BaseOutputParser):
    """将输出解析为逗号分隔的列表"""

    def parse(self, text: str) -> list[str]:
        return [item.strip() for item in text.strip().split(",")]

    @property
    def _type(self) -> str:
        return "comma_separated"

prompt = ChatPromptTemplate.from_template(
    "列出5种{category}，用逗号分隔，不要编号："
)

chain = prompt | llm | CommaSeparatedParser()
result = chain.invoke({"category": "编程语言"})
print(result)  # ['Python', 'JavaScript', 'Java', 'Go', 'Rust']

6. Chain 链式调用

6.1 LLMChain（基础链）

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

prompt = ChatPromptTemplate.from_template("用一句话解释：{concept}")
llm = ChatOpenAI(model="gpt-4o-mini")

# 使用 LCEL（LangChain Expression Language）管道语法
chain = prompt | llm | StrOutputParser()

result = chain.invoke({"concept": "微服务架构"})
print(result)

6.2 SequentialChain（顺序链）

# 场景：先生成文章大纲，再根据大纲撰写正文

outline_prompt = ChatPromptTemplate.from_template(
    "请为以下主题生成一个5点的文章大纲：\n主题：{topic}"
)

write_prompt = ChatPromptTemplate.from_template(
    "根据以下大纲，撰写一篇500字的文章：\n大纲：{outline}"
)

# 方式一：使用 RunnablePassthrough 传递中间结果
from langchain_core.runnables import RunnablePassthrough

chain = (
    {"outline": outline_prompt | llm | StrOutputParser()}
    | RunnablePassthrough.assign(
        article=lambda x: (write_prompt | llm | StrOutputParser()).invoke(
            {"outline": x["outline"]}
        )
    )
)

result = chain.invoke({"topic": "人工智能在医疗领域的应用"})
print("大纲：", result["outline"])
print("文章：", result["article"])

6.3 并行执行与分支

from langchain_core.runnables import RunnableParallel

# 同时生成英文和中文版本
en_prompt = ChatPromptTemplate.from_template("Translate to English: {text}")
zh_prompt = ChatPromptTemplate.from_template("翻译成中文：{text}")

parallel_chain = RunnableParallel(
    english=en_prompt | llm | StrOutputParser(),
    chinese=zh_prompt | llm | StrOutputParser(),
)

result = parallel_chain.invoke({"text": "天道酬勤"})
print(result["english"])  # Heaven rewards diligence
print(result["chinese"])  # 天道酬勤

7. Agent 智能体

7.1 ReAct Agent

ReAct（Reasoning + Acting）是最经典的 Agent 模式：LLM 先推理（Thought），再行动（Action），观察结果（Observation），循环直到完成任务。

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.tools import tool
from langchain_core.prompts import PromptTemplate

@tool
def search_weather(city: str) -> str:
    """查询指定城市的天气信息"""
    # 实际项目中接入天气 API
    weather_data = {"北京": "晴，25°C", "上海": "多云，22°C"}
    return weather_data.get(city, f"未找到{city}的天气数据")

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

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

react_prompt = PromptTemplate.from_template("""
Answer the following questions as best you can. You have access to the following tools:

{tools}

Use the following format:

Question: the input question
Thought: reasoning about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Begin!

Question: {input}
Thought:{agent_scratchpad}
""")

agent = create_react_agent(llm, [search_weather, calculate], react_prompt)
agent_executor = AgentExecutor(agent=agent, tools=[search_weather, calculate], verbose=True)

result = agent_executor.invoke({"input": "北京今天多少度？如果升温5度是多少？"})
print(result["output"])

7.2 使用 LangGraph 构建现代 Agent

LangGraph 是 LangChain 团队推出的图编排框架，比传统 AgentExecutor 更灵活：

from langgraph.prebuilt import create_react_agent

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

# 一行代码创建 ReAct Agent
agent = create_react_agent(llm, tools=[search_weather, calculate])

result = agent.invoke({"messages": [("human", "北京天气怎么样？")]})
for msg in result["messages"]:
    print(f"{msg.type}: {msg.content}")

8. Tool 工具系统开发

8.1 使用 @tool 装饰器

from langchain_core.tools import tool
from typing import Optional

@tool
def get_stock_price(symbol: str, market: str = "A股") -> str:
    """获取股票的当前价格。

    Args:
        symbol: 股票代码，如 '600519'
        market: 市场类型，A股或美股
    """
    # 模拟数据
    prices = {"600519": 1800.00, "AAPL": 195.50}
    price = prices.get(symbol, "未知")
    return f"{symbol} 当前价格：{price}"

@tool
def send_notification(
    recipient: str,
    message: str,
    priority: Optional[str] = "normal"
) -> str:
    """发送通知消息给指定用户。

    Args:
        recipient: 接收者ID或邮箱
        message: 通知内容
        priority: 优先级，可选 low/normal/high
    """
    return f"已发送 [{priority}] 通知给 {recipient}: {message}"

8.2 StructuredTool（高级定义）

from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(description="搜索关键词")
    max_results: int = Field(default=5, description="最大返回结果数")
    language: str = Field(default="zh", description="搜索语言")

def search_web(query: str, max_results: int = 5, language: str = "zh") -> str:
    """模拟网络搜索"""
    return f"找到 {max_results} 条关于 '{query}' 的结果"

search_tool = StructuredTool.from_function(
    func=search_web,
    name="web_search",
    description="在互联网上搜索信息",
    args_schema=SearchInput,
)

8.3 工具包（Toolkit）

from langchain_core.tools import BaseTool

class MathToolkit:
    """数学计算工具集"""

    def get_tools(self) -> list[BaseTool]:
        return [
            tool(self.add),
            tool(self.multiply),
            tool(self.power),
        ]

    def add(self, a: float, b: float) -> float:
        """两数相加"""
        return a + b

    def multiply(self, a: float, b: float) -> float:
        """两数相乘"""
        return a * b

    def power(self, base: float, exponent: float) -> float:
        """幂运算"""
        return base ** exponent

9. Memory 记忆管理

9.1 ConversationBufferMemory（完整记忆）

from langchain.memory import ConversationBufferMemory
from langchain_openai import ChatOpenAI
from langchain.chains import ConversationChain

memory = ConversationBufferMemory(return_messages=True)

llm = ChatOpenAI(model="gpt-4o-mini")
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True,
)

# 多轮对话
conversation.invoke({"input": "我叫张三"})
conversation.invoke({"input": "我是一名程序员"})
conversation.invoke({"input": "你还记得我的名字和职业吗？"})
# 模型会记住之前的信息

9.2 ConversationSummaryMemory（摘要记忆）

from langchain.memory import ConversationSummaryMemory

# 当对话很长时，自动压缩为摘要
memory = ConversationSummaryMemory(
    llm=ChatOpenAI(model="gpt-4o-mini"),
    return_messages=True,
)

conversation = ConversationChain(llm=llm, memory=memory)
# 适合长对话场景，自动管理 token 用量

9.3 在 LangGraph 中管理状态

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

# MessagesState 内置了 messages 列表，天然支持对话记忆
def agent_node(state: MessagesState):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

graph = StateGraph(MessagesState)
graph.add_node("agent", agent_node)
graph.add_edge(START, "agent")
graph.add_edge("agent", END)

app = graph.compile()

# 每次调用都传入完整的历史消息
result = app.invoke({"messages": [("human", "你好")]})

10. LangGraph 核心概念与 StateGraph

10.1 什么是 LangGraph

LangGraph 是基于图的 AI 应用编排框架。与传统的线性 Chain 不同，它允许你构建：

有环图：支持循环和条件分支
有状态的工作流：自动管理状态传递
人机协作：支持 Human-in-the-Loop
持久化：支持检查点和断点续传

10.2 StateGraph 基础

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

# 定义状态结构
class WorkflowState(TypedDict):
    # 使用 Annotated 和 operator.add 实现"追加"而非"覆盖"
    messages: Annotated[list, operator.add]
    step_count: int
    status: str

# 创建图
graph = StateGraph(WorkflowState)

# 定义节点函数
def step_one(state: WorkflowState) -> dict:
    return {
        "messages": ["步骤1完成"],
        "step_count": state["step_count"] + 1,
    }

def step_two(state: WorkflowState) -> dict:
    return {
        "messages": ["步骤2完成"],
        "step_count": state["step_count"] + 1,
        "status": "done",
    }

# 添加节点和边
graph.add_node("step_one", step_one)
graph.add_node("step_two", step_two)
graph.add_edge(START, "step_one")
graph.add_edge("step_one", "step_two")
graph.add_edge("step_two", END)

# 编译并运行
app = graph.compile()
result = app.invoke({"messages": [], "step_count": 0, "status": "pending"})
print(result)
# {'messages': ['步骤1完成', '步骤2完成'], 'step_count': 2, 'status': 'done'}

11. LangGraph 节点与边的定义

11.1 条件边（Conditional Edges）

def should_continue(state: WorkflowState) -> str:
    """根据状态决定下一步"""
    if state["step_count"] >= 3:
        return "finish"
    return "continue"

def process_step(state: WorkflowState) -> dict:
    return {"step_count": state["step_count"] + 1, "messages": [f"处理第{state['step_count']+1}步"]}

def finish_node(state: WorkflowState) -> dict:
    return {"status": "completed", "messages": ["全部完成！"]}

graph = StateGraph(WorkflowState)
graph.add_node("process", process_step)
graph.add_node("finish", finish_node)

graph.add_edge(START, "process")
# 条件边：根据返回值跳转到不同节点
graph.add_conditional_edges(
    "process",
    should_continue,
    {"continue": "process", "finish": "finish"},
)
graph.add_edge("finish", END)

app = graph.compile()
result = app.invoke({"messages": [], "step_count": 0, "status": "pending"})

11.2 带工具调用的 Agent 图

from langgraph.prebuilt import ToolNode
from langchain_core.messages import HumanMessage

@tool
def lookup_database(query: str) -> str:
    """查询数据库获取信息"""
    return f"数据库查询结果：关于'{query}'的数据"

tools = [lookup_database]
tool_node = ToolNode(tools)

def agent(state: MessagesState):
    response = llm.bind_tools(tools).invoke(state["messages"])
    return {"messages": [response]}

def should_use_tool(state: MessagesState):
    last_msg = state["messages"][-1]
    if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
        return "tools"
    return "end"

graph = StateGraph(MessagesState)
graph.add_node("agent", agent)
graph.add_node("tools", tool_node)

graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_use_tool, {"tools": "tools", "end": END})
graph.add_edge("tools", "agent")  # 工具执行完后回到 agent

app = graph.compile()
result = app.invoke({"messages": [HumanMessage(content="查询今天的销售数据")]})

12. 多 Agent 协作模式

12.1 Supervisor 模式

from typing import Literal

# 定义各专家 Agent
def researcher_agent(state: MessagesState):
    """研究员：负责信息收集"""
    prompt = "你是研究员，负责收集和整理信息。"
    response = llm.invoke([("system", prompt)] + state["messages"])
    return {"messages": [response]}

def writer_agent(state: MessagesState):
    """撰稿人：负责内容撰写"""
    prompt = "你是撰稿人，根据提供的素材撰写文章。"
    response = llm.invoke([("system", prompt)] + state["messages"])
    return {"messages": [response]}

def supervisor_agent(state: MessagesState) -> dict:
    """主管：决定下一步由谁执行"""
    prompt = """你是项目主管。根据当前进展，决定下一步：
    - 如果需要更多信息，选择 'researcher'
    - 如果信息足够可以撰写，选择 'writer'
    - 如果任务完成，选择 'FINISH'
    """
    response = llm.invoke([("system", prompt)] + state["messages"])
    # 解析响应决定路由
    content = response.content.lower()
    if "writer" in content:
        next_step = "writer"
    elif "finish" in content:
        next_step = "FINISH"
    else:
        next_step = "researcher"
    return {"next": next_step, "messages": [response]}

# 构建 Supervisor 图
from typing import TypedDict

class SupervisorState(TypedDict):
    messages: Annotated[list, operator.add]
    next: str

graph = StateGraph(SupervisorState)
graph.add_node("supervisor", supervisor_agent)
graph.add_node("researcher", researcher_agent)
graph.add_node("writer", writer_agent)

graph.add_edge(START, "supervisor")
graph.add_conditional_edges(
    "supervisor",
    lambda x: x["next"],
    {"researcher": "researcher", "writer": "writer", "FINISH": END},
)
graph.add_edge("researcher", "supervisor")
graph.add_edge("writer", "supervisor")

app = graph.compile()

12.2 并行 Agent 执行

from langgraph.graph import StateGraph

def parallel_agents(state: dict):
    """多个 Agent 并行处理不同子任务"""
    # 可以使用 asyncio.gather 并行调用多个 Agent
    import asyncio

    async def run_agents():
        tasks = [
            agent_ainvoke(state, "agent_a"),
            agent_ainvoke(state, "agent_b"),
        ]
        return await asyncio.gather(*tasks)

    results = asyncio.run(run_agents())
    return {"results": results}

13. RAG 集成实战

13.1 基础 RAG 链

from langchain_community.document_loaders import TextLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

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

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

# 3. 创建向量数据库
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(chunks, embeddings, persist_directory="./chroma_db")
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# 4. 构建 RAG 链
rag_prompt = ChatPromptTemplate.from_template("""
基于以下上下文回答问题。如果上下文中没有相关信息，请说明你不确定。

上下文：
{context}

问题：{question}
""")

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | llm
    | StrOutputParser()
)

result = rag_chain.invoke("公司的请假制度是什么？")
print(result)

14. 流式输出处理

14.1 基础流式输出

# 方法一：stream 方法
chain = prompt | llm | StrOutputParser()

for chunk in chain.stream({"concept": "人工智能"}):
    print(chunk, end="", flush=True)

14.2 在 LangGraph 中流式输出

# 流式输出 Agent 的思考过程
app = graph.compile()

for event in app.stream({"messages": [("human", "分析今天的市场走势")]},
                         stream_mode="updates"):
    for node, output in event.items():
        print(f"[{node}]", output)
        print("---")

14.3 异步流式

import asyncio

async def stream_response():
    async for chunk in chain.astream({"concept": "量子计算"}):
        print(chunk, end="", flush=True)

asyncio.run(stream_response())

15. 生产部署最佳实践

15.1 错误处理与重试

from langchain_core.runnables import RunnableLambda
import time

def robust_llm_call(inputs):
    """带重试的 LLM 调用"""
    max_retries = 3
    for attempt in range(max_retries):
        try:
            return llm.invoke(inputs)
        except Exception as e:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt
                print(f"重试 {attempt + 1}/{max_retries}，等待 {wait_time}s: {e}")
                time.sleep(wait_time)
            else:
                raise

robust_chain = prompt | RunnableLambda(robust_llm_call) | StrOutputParser()

15.2 链路追踪（LangSmith）

import os

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

# 所有调用自动上报到 LangSmith 平台
# 可以在 https://smith.langchain.com 查看调用链路

15.3 缓存优化

from langchain_core.globals import set_llm_cache
from langchain_community.cache import SQLiteCache

# 启用语义缓存，避免重复调用
set_llm_cache(SQLiteCache(database_path=".langchain.db"))

# 相同的输入会直接返回缓存结果
result1 = llm.invoke("什么是机器学习？")  # 实际调用
result2 = llm.invoke("什么是机器学习？")  # 从缓存返回

16. 实战项目：智能客服系统

下面我们构建一个完整的智能客服系统，整合本教程所学的所有核心概念。

16.1 系统架构

用户消息 → 意图识别 → 路由分发
                        ├── 订单查询 Agent（调用数据库工具）
                        ├── 产品咨询 Agent（RAG 检索产品文档）
                        ├── 投诉处理 Agent（需人工确认）
                        └── 通用问答 Agent

16.2 完整实现

"""智能客服系统 - 基于 LangGraph 的多 Agent 架构"""
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from langchain_core.tools import tool
import operator

# ==================== 1. 定义工具 ====================

@tool
def query_order(order_id: str) -> str:
    """根据订单号查询订单状态和详情"""
    # 模拟数据库查询
    orders = {
        "ORD001": {"status": "已发货", "tracking": "SF1234567890", "items": "iPhone 15"},
        "ORD002": {"status": "待付款", "amount": "5999元", "items": "MacBook Air"},
        "ORD003": {"status": "已完成", "items": "AirPods Pro"},
    }
    order = orders.get(order_id)
    if order:
        return f"订单 {order_id}: {order}"
    return f"未找到订单 {order_id}，请确认订单号是否正确"

@tool
def search_product_info(query: str) -> str:
    """搜索产品信息和常见问题"""
    # 实际项目中对接 RAG 系统
    docs = {
        "退换货": "退换货政策：7天无理由退换，15天质量问题换货。请保留原始包装。",
        "保修": "所有产品享受1年质保，Apple Care+可延长至2年。",
        "配送": "标准配送3-5个工作日，加急配送次日达（需额外收费）。",
    }
    for key, value in docs.items():
        if key in query:
            return value
    return f"未找到关于'{query}'的产品信息"

@tool
def escalate_to_human(reason: str) -> str:
    """将对话升级给人工客服"""
    return f"已转接人工客服。原因：{reason}。请稍候，客服代表将很快接入。"

# ==================== 2. 定义状态 ====================

class CustomerServiceState(TypedDict):
    messages: Annotated[list, operator.add]
    intent: str
    requires_human: bool
    conversation_id: str

# ==================== 3. 定义节点 ====================

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

def classify_intent(state: CustomerServiceState) -> dict:
    """意图识别节点"""
    last_msg = state["messages"][-1].content

    classification_prompt = f"""分析用户消息的意图，返回以下类别之一：
    - order_query: 查询订单状态、物流信息
    - product_info: 产品咨询、退换货、保修
    - complaint: 投诉、不满、要求赔偿
    - general: 一般问候、闲聊、其他

    用户消息：{last_msg}
    意图："""

    response = llm.invoke([HumanMessage(content=classification_prompt)])
    intent = response.content.strip().lower()

    # 标准化意图
    for valid_intent in ["order_query", "product_info", "complaint", "general"]:
        if valid_intent in intent:
            intent = valid_intent
            break
    else:
        intent = "general"

    return {"intent": intent}

def route_intent(state: CustomerServiceState) -> str:
    """路由分发"""
    intent = state.get("intent", "general")
    if intent == "order_query":
        return "order_agent"
    elif intent == "product_info":
        return "product_agent"
    elif intent == "complaint":
        return "complaint_agent"
    return "general_agent"

def order_agent(state: CustomerServiceState) -> dict:
    """订单查询 Agent"""
    system = SystemMessage(content="""你是订单查询专员。帮助用户查询订单状态。
    使用 query_order 工具查询订单。如果用户没提供订单号，请礼貌询问。""")

    tools = [query_order]
    agent = create_react_agent(llm, tools)
    result = agent.invoke({"messages": [system] + state["messages"]})

    return {"messages": [AIMessage(content=result["messages"][-1].content)]}

def product_agent(state: CustomerServiceState) -> dict:
    """产品咨询 Agent"""
    system = SystemMessage(content="""你是产品顾问。回答用户关于产品的问题。
    使用 search_product_info 工具搜索相关信息。""")

    tools = [search_product_info]
    agent = create_react_agent(llm, tools)
    result = agent.invoke({"messages": [system] + state["messages"]})

    return {"messages": [AIMessage(content=result["messages"][-1].content)]}

def complaint_agent(state: CustomerServiceState) -> dict:
    """投诉处理 Agent"""
    system = SystemMessage(content="""你是投诉处理专员。
    1. 先表达歉意和理解
    2. 了解具体问题
    3. 提供解决方案
    4. 如果用户情绪激动或要求赔偿，使用 escalate_to_human 转人工""")

    tools = [escalate_to_human]
    agent = create_react_agent(llm, tools)
    result = agent.invoke({"messages": [system] + state["messages"]})

    return {"messages": [AIMessage(content=result["messages"][-1].content)]}

def general_agent(state: CustomerServiceState) -> dict:
    """通用问答 Agent"""
    system = SystemMessage(content="你是一位友好的客服助手，回答用户的一般性问题。")
    response = llm.invoke([system] + state["messages"])
    return {"messages": [response]}

# ==================== 4. 构建图 ====================

graph = StateGraph(CustomerServiceState)

# 添加节点
graph.add_node("classify", classify_intent)
graph.add_node("order_agent", order_agent)
graph.add_node("product_agent", product_agent)
graph.add_node("complaint_agent", complaint_agent)
graph.add_node("general_agent", general_agent)

# 添加边
graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_intent, {
    "order_agent": "order_agent",
    "product_agent": "product_agent",
    "complaint_agent": "complaint_agent",
    "general_agent": "general_agent",
})
graph.add_edge("order_agent", END)
graph.add_edge("product_agent", END)
graph.add_edge("complaint_agent", END)
graph.add_edge("general_agent", END)

# 使用 Memory 实现多轮对话
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

# ==================== 5. 运行系统 ====================

def chat(user_input: str, thread_id: str = "session-001"):
    """客服对话入口"""
    config = {"configurable": {"thread_id": thread_id}}

    result = app.invoke(
        {
            "messages": [HumanMessage(content=user_input)],
            "intent": "",
            "requires_human": False,
            "conversation_id": thread_id,
        },
        config=config,
    )

    return result["messages"][-1].content

# 测试对话
print("=== 订单查询 ===")
print(chat("我想查一下订单 ORD001 的状态"))

print("\n=== 产品咨询 ===")
print(chat("你们的退换货政策是什么？"))

print("\n=== 投诉处理 ===")
print(chat("我的手机屏幕碎了，你们质量太差了！我要投诉！"))

print("\n=== 多轮对话 ===")
print(chat("刚才说的订单 ORD001 大概什么时候到？", thread_id="session-001"))

16.3 部署为 API 服务

"""使用 FastAPI 部署客服系统"""
from fastapi import FastAPI
from pydantic import BaseModel
import uvicorn

app_fastapi = FastAPI(title="智能客服 API")

class ChatRequest(BaseModel):
    message: str
    session_id: str = "default"

class ChatResponse(BaseModel):
    reply: str
    intent: str
    session_id: str

@app_fastapi.post("/chat", response_model=ChatResponse)
async def chat_endpoint(request: ChatRequest):
    result = app.invoke(
        {
            "messages": [HumanMessage(content=request.message)],
            "intent": "",
            "requires_human": False,
            "conversation_id": request.session_id,
        },
        config={"configurable": {"thread_id": request.session_id}},
    )

    return ChatResponse(
        reply=result["messages"][-1].content,
        intent=result.get("intent", "unknown"),
        session_id=request.session_id,
    )

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

总结

本教程从基础到实战，系统性地介绍了 LangChain 和 LangGraph 的核心能力：

模块	核心能力	适用场景
Prompt Template	标准化提示词管理	所有 LLM 应用
Chain	线性工作流	简单的多步处理
Agent	自主决策与工具调用	复杂的动态任务
Memory	对话上下文管理	多轮对话场景
LangGraph	图编排与状态管理	复杂工作流、多 Agent 协作

下一步学习建议：

阅读 LangChain 官方文档深入了解更多组件
探索 LangGraph 官方教程学习高级模式
使用 LangSmith 进行链路追踪和调试
结合 RAG 构建知识增强型应用（参见本系列 RAG 教程）

📅 最后更新：2026年5月 | 📝 适用版本：LangChain ≥ 0.3、LangGraph ≥ 0.2

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

教程简介