LangChain 高级概念

LangChain 的高级概念围绕多智能体协作、知识增强检索、外部系统集成和工程化观测展开，构建生产级 LLM 应用。

排序说明（常用→不常用）

顺序	概念	重要程度
1	Multi-agent 多智能体	高，复杂任务的核心拆解手段
2	RAG 检索增强生成	高，知识密集型应用的基础
3	Structured Output 结构化输出	高，工程落地的必备能力
4	MCP 模型上下文协议	中，外部工具集成标准
5	Context Engineering 上下文工程	中，提示词优化的系统性方法
6	Observability 可观测性	中，生产运维必备

---

一、Multi-agent（多智能体）—— 复杂任务的核心

多个 Agent 协作分工，每个专注自己的领域，适合复杂任务拆解。

1.1 协作模式

模式	说明	适用场景
Handoffs（交接）	一个 Agent 完成当前任务后转交给下一个	顺序流程，每步不同专业
Router（路由）	LLM 根据输入决定交给哪个 Agent	分类/分发任务
Custom Workflow（自定义工作流）	固定图结构定义 Agent 协作	复杂多步骤流程
Subagents（子智能体）	一个 Agent 可调用其他 Agent	层次化任务分解

1.2 Handoffs 交接模式

from langchain.agents import create_agent
from langchain.agents.output import Handoff

def create_sales_agent():
    return create_agent(
        model=model,
        tools=[...],
        system_prompt="你是一个专业销售"
    )

def create_support_agent():
    return create_agent(
        model=model,
        tools=[...],
        system_prompt="你是一个技术支持"
    )

# 交接：当销售无法回答时转给支持
handoff = Handoff(agent_name="support_agent")


"""例子"""
from langgraph import Graph, AgentNode, HandoffNode

# 创建 Agent 节点
sales_agent = AgentNode(name="sales_agent", model=model, system_prompt="你是专业销售")
support_agent = AgentNode(name="support_agent", model=model, system_prompt="你是技术支持")

# 创建 Handoff 节点
handoff_node = HandoffNode(name="handoff_to_support", target_agent="support_agent")

# 构建图
graph = Graph()
graph.add_nodes([sales_agent, support_agent, handoff_node])

# 连接节点
graph.connect(sales_agent, handoff_node)  # 销售节点输出 Handoff 时走到 handoff_node
graph.connect(handoff_node, support_agent)  # Handoff 指向支持节点

1.3 Router 路由模式

from langchain.agents import create_router_agent

def route_fn(input_text: str) -> str:
    if "buy" in input_text.lower() or "price" in input_text.lower():
        return "sales_agent"
    elif "error" in input_text.lower() or "bug" in input_text.lower():
        return "support_agent"
    return "general_agent"

router = create_router_agent(
    model=model,
    agents=[sales_agent, support_agent, general_agent],
    router_fn=route_fn
)

1.4 自定义工作流

from langgraph.graph import StateGraph, START, END

class MultiAgentState(TypedDict):
    messages: list
    current_agent: str
    task_result: dict

# 构建工作流图
builder = StateGraph(MultiAgentState)
builder.add_node("researcher", research_agent)
builder.add_node("writer", writer_agent)
builder.add_node("reviewer", review_agent)

builder.add_edge(START, "researcher")
builder.add_edge("researcher", "writer")
builder.add_edge("writer", "reviewer")
builder.add_edge("reviewer", END)

---

二、RAG（检索增强生成）—— 知识密集型应用的基础

RAG = Retrieval Augmented Generation，让 LLM 基于私有知识库回答问题。

2.1 RAG 流程

用户问题 → 检索（Query） → 找到相关文档 → 拼入提示词 → LLM 生成回答

2.2 核心组件

组件	作用	常用实现
Document Loader	加载各种格式文档	PDF、Markdown、HTML
Text Splitter	将长文档切分成块	按字数、语义切分
Embeddings	将文本转为向量	OpenAI、Azure OpenAI、HuggingFace
Vector Store	存储向量 + 索引	Pinecone、Chroma、FAISS
Retriever	根据 Query 检索相关文档	相似度搜索、BM25

2.3 完整 RAG 示例

from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chat_models import init_chat_model
from langchain.chains import RetrievalQA

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

# 2. 切分文本
splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
chunks = splitter.split_documents(documents)

# 3. 生成嵌入并存入向量库
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)

# 4. 构建 RAG 链
retriever = vectorstore.as_retriever()
qa_chain = RetrievalQA.from_chain_type(
    llm=init_chat_model("gpt-4"),
    chain_type="stuff",
    retriever=retriever
)

# 5. 问答
result = qa_chain.invoke({"query": "某问题"})

2.4 RAG 优化策略

策略	说明
Query 改写	用 LLM 改写用户问题，提升检索效果
HyDE	让 LLM 先生成假设答案，再检索相似文档
重排序（Rerank）	检索后用模型重排序，提升相关性
混合检索	结合向量搜索 + 关键词搜索（BM25）
元数据过滤	按时间、来源等条件过滤检索结果

2.5 检索模式对比

模式	说明	适用场景
similarity_search	向量相似度搜索	语义匹配
mmr (最大边际相关性)	兼顾相关性和多样性	避免重复结果
similarity_score_threshold	设置相似度阈值	控制质量下限

---

三、Structured Output（结构化输出）—— 工程落地的必备

让 LLM 输出符合特定 Schema 的结构化数据，便于程序处理。

3.1 基本用法：with_structured_output

from pydantic import BaseModel
from langchain.chat_models import init_chat_model

class Joke(BaseModel):
    setup: str
    punchline: str
    rating: int  # 1-10

model = init_chat_model("gpt-4")
joke_model = model.with_structured_output(Joke)

joke = joke_model.invoke("给我讲个程序员笑话")
# → Joke(setup="为什么程序员总是分不清万圣节和圣诞节？", punchline="因为 Oct 31 = Dec 25", rating=8)

3.2 带描述的 Schema

from pydantic import BaseModel, Field

class WeatherResponse(BaseModel):
    city: str = Field(description="城市名称")
    temperature: int = Field(description="温度（摄氏度）")
    condition: str = Field(description="天气状况，如'晴'、'雨'")
    humidity: int = Field(description="湿度百分比", ge=0, le=100)

3.3 可选字段和列表

class ResearchResults(BaseModel):
    topic: str
    summary: str
    key_points: list[str] = Field(default_factory=list)
    sources: list[dict] | None = None  # 可选
    confidence: float = 1.0

3.4 JSON Mode vs Structured Output

特性	JSON Mode	Structured Output
语法保证	无，可能有 JSON 语法错误	有，Pydantic 验证
类型验证	无	有
速度	略快	略慢（多一次验证）
灵活性	高	低（受 Schema 限制）

---

四、MCP（模型上下文协议）—— 外部工具集成标准

MCP = Model Context Protocol，Anthropic 推出的让 AI 模型与外部工具/数据源交互的标准协议。

4.1 MCP 架构

LLM ←→ MCP Client ←→ MCP Server ←→ 外部工具/数据源

4.2 MCP 工具定义

from langchain.tools import MCPTool

# 定义一个 MCP 工具
mcp_tool = MCPTool(
    name="filesystem_read",
    description="读取文件内容",
    server_name="filesystem",
    server_uri="http://localhost:3000"
)

4.3 MCP 服务器示例

# 服务器端（Node.js / Python）
{
    "name": "filesystem",
    "tools": [
        {
            "name": "read_file",
            "description": "读取文件",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "path": {"type": "string"}
                }
            }
        }
    ]
}

4.4 在 LangChain 中使用

from langchain.agents import create_agent
from langchain.tools.mcp import MCPToolPool

# 连接 MCP 服务器
mcp_pool = MCPToolPool(["http://localhost:3000"])

agent = create_agent(
    model=model,
    tools=mcp_pool.get_tools(),
    system_prompt="你是一个文件操作助手"
)

---

五、Context Engineering（上下文工程）—— 提示词优化的系统性方法

上下文工程是系统性地设计、管理输入上下文的技术，最大化 LLM 输出质量。

5.1 核心原则

原则	说明
Relevance	只包含与当前任务相关的信息
Density	压缩冗余信息，提取关键内容
Ordering	重要信息放在开头或结尾（位置效应）
Format	结构化格式比纯文本更易被理解

5.2 上下文优化技术

技术	说明
摘要	对长文档进行摘要，保留关键信息
窗口滑动	滑动窗口选择最相关的上下文片段
分层检索	先粗筛再精筛，逐步缩小范围
动态注入	根据任务类型动态选择提示词模板

5.3 Few-shot Prompting

from langchain.prompts import FewShotPromptTemplate, PromptTemplate

examples = [
    {"input": "3 * 4", "output": "12"},
    {"input": "5 + 7", "output": "12"},
]

prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=PromptTemplate.from_template("输入: {input}\n输出: {output}"),
    suffix="输入: {num1} * {num2}\n输出:",
    input_variables=["num1", "num2"]
)

---

六、Observability（可观测性）—— 生产运维必备

可观测性让你了解 LLM 应用在生产环境中的行为，便于调试和优化。

6.1 LangSmith 集成

from langsmith import traceable
from langchain.chat_models import init_chat_model

# 自动追踪
@traceable
def my_agent_function(input_text: str):
    model = init_chat_model("gpt-4")
    return model.invoke(input_text)

6.2 追踪内容

追踪项	说明
输入/输出	每次 LLM 调用的请求和响应
Token 消耗	使用的 token 数量和成本
延迟	端到端延迟和各阶段延迟
工具调用	工具名称、参数、执行结果
自定义指标	业务相关的自定义指标

6.3 常见可观测性指标

指标	说明
首次响应时间（TTFT）	用户看到首个 token 的时间
总响应时间	完整响应的端到端延迟
Token 速度	每秒输出的 token 数
错误率	LLM 调用失败的比例
工具调用成功率	工具执行的成功/失败比例

6.4 生产环境检查清单

• 集成 LangSmith 或其他可观测性工具
• 设置成本告警（超预算时通知）
• 记录每次交互的输入/输出
• 追踪关键业务指标
• 设置重试和降级策略

---

简单总结

概念	核心作用	关键 API/组件
Multi-agent	多 Agent 协作分工	Handoffs、Router、Custom Workflow
RAG	基于私有知识库回答	Loader、Splitter、Embeddings、VectorStore、Retriever
Structured Output	结构化输出	with_structured_output()、Pydantic
MCP	外部工具集成标准	MCPTool、MCPToolPool
Context Engineering	上下文优化	Few-shot、摘要、动态注入
Observability	生产运维监控	LangSmith、traceable、指标追踪

使用频率排序（从高到低）：
Structured Output → RAG → Multi-agent → Context Engineering → MCP → Observability