LangChain

🧱 1. LangChain 1.0 的四大核心组件

LangChain 1.0 的使命是：用搭积木的方式，把 LLM 调用、工具、记忆这些零散的能力组装成可运行的应用。 它的设计围绕四个抽象展开。

① Chain（链）

本质：把多个步骤串起来，定义一个确定的执行流程。 你可以像写配置文件一样，声明“先用这个 prompt 调 LLM，再把输出传给下一个 prompt，或者传给某个 Python 函数”。最常见的 LLMChain 就是 PromptTemplate + LLM + 可选的输出解析器 的组合。

from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI
from langchain.chains import LLMChain

prompt = PromptTemplate.from_template("给一家卖{product}的公司起三个名字")
chain = LLMChain(llm=OpenAI(), prompt=prompt)
print(chain.run("智能手表"))

Chain 让“多次 LLM 调用 + 数据处理”变成一个可复用的管道。

② Agent（智能体）

本质：让 LLM 自己决定“下一步做什么”。

它不按固定顺序执行，而是用 LLM 作为推理引擎，在一个循环里重复“思考 → 选择工具 → 执行 → 观察结果”，直到认为可以回答用户。Agent 的核心是决策，而不是死板的流程。

from langchain.agents import load_tools, initialize_agent, AgentType

tools = load_tools(["serpapi", "llm-math"], llm=llm)
agent = initialize_agent(tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True)
agent.run("马斯克几岁了？他的年龄的平方根是多少？")

这段代码背后，Agent 会先搜索“马斯克年龄”，然后用计算器开平方，最后整合回答。

③ Tool（工具）

本质：LLM 的外部手和脚。

任何被包装成“名字 + 描述 + 参数 schema + 可调用函数”的东西都可以是 Tool。Agent 通过 Tool 的 description 来决定什么时候用它、传什么参数。搜索、计算器、数据库查询、API 调用，都是 Tool。

from langchain.tools import Tool

def get_weather(city: str) -> str:
    return f"{city}：晴，25°C"

weather_tool = Tool(
    name="Weather",
    func=get_weather,
    description="输入城市名，返回天气。参数 city 为字符串。"
)

④ Memory（记忆）

本质：跨轮次保持的对话状态。

LLM 本身是无状态的，每次调用只看当前 prompt。Memory 把历史对话、实体、摘要存起来，下次请求时自动注入到 prompt 的合适位置，让 Chatbot 能“记住”之前说过的话。

from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory(return_messages=True)
memory.chat_memory.add_user_message("我叫小明")
memory.chat_memory.add_ai_message("你好小明！")
# 在 Chain 中集成 memory 后，LLM 会看到完整历史

这四个组件，构成了 LangChain 1.0 的世界观：Chain 管流程，Agent 管决策，Tool 管能力，Memory 管状态。

---

⚡ 2. LangChain 2.0 的 LCEL 是什么？和 1.0 的 Chain 有什么本质区别？

LCEL（LangChain Expression Language） 是一种用管道符 | 把组件串起来的声明式语言。它让链的定义从“黑箱函数调用”变成“数据流管道”。

# LCEL 写法
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template("给一家卖{product}的公司起名字")
model = ChatOpenAI()
chain = prompt | model | StrOutputParser()
print(chain.invoke({"product": "智能手表"}))

LCEL 和 1.0 Chain 的本质区别，一句话：

• 1.0 Chain 是“封装好的类”：你需要继承 Chain 并实现 _call 方法，流程被封装在类内部，修改、组合、调试都不透明。

• 2.0 LCEL 是“数据流管道”：每个步骤都是独立的 Runnable 对象，用 | 连接。输入数据像流水一样从左流到右，每一步的输入输出完全可见、可测试、可组合。

用一个图表对比它们的设计哲学：

1.0 Chain：
  输入 → [黑箱 LLMChain] → 输出
         内部: prompt→LLM→parser，但外部不可见

2.0 LCEL：
  输入 → prompt → model → parser → 输出
         ↑       ↑       ↑
      每个节点都可以被单独检查、修改、替换

五个根本差异：

查看内嵌表格

一个体现 LCEL 强大组合性的代码例子：

# 把两个链并联：一个写大纲，一个写正文，然后合并
outline_chain = prompt_outline | model | StrOutputParser()
essay_chain = prompt_essay | model | StrOutputParser()
combined = RunnableParallel(outline=outline_chain, essay=essay_chain)
result = combined.invoke({"topic": "AI"})
# result 是 {"outline": "...", "essay": "..."}

在 1.0 中要实现这种“并联 + 合并”，你得上 SequentialChain 配 SimpleMemory，代码量翻倍且极难调试。LCEL 让它变得像搭乐高。

---

🔄 3. LangChain 1.0 到 2.0 的架构演进解决了哪些核心痛点？

1.0 的成功验证了“LLM 应用框架”的需求，但也暴露了三个致命伤。2.0 的每一次重构，都在对症下药。

痛点一：不透明与难调试 1.0 的 Chain 内部像一个黑箱。你想看中间某一步的输出？得加一堆 verbose=True 看日志。出了错，回溯困难。 解决方案：LCEL 的可观测性。每个 Runnable 都有自己的回调钩子，LangSmith 集成了追踪。链的每一步输入输出都被自动记录，出问题时一眼就能看到是哪个节点崩了。

痛点二：组合性差，自定义受限 1.0 有很多“专用 Chain”：LLMChain、SequentialChain、ConversationChain……各自有不同的调用方式和限制。如果你需要“先并行查三个数据，再汇总”，很难用现有 Chain 组合出来。 解决方案：Runnable 原子化。所有东西都是 Runnable，输入输出统一。RunnableParallel、RunnableBranch、RunnableLambda 这些基础积木，让你可以自由构建任意拓扑的流程，而不必被框架预设的 Chain 类型框死。

痛点三：流式、异步、批处理支持参差不齐 1.0 的很多 Chain 没有实现 _astream 或 _acall，导致在异步 Web 服务或流式输出场景下完全不能用。 解决方案：LCEL 从底层强制统一了 invoke、ainvoke、stream、batch 等调用接口。任何 LCEL 链，不用改一行内部代码，就能直接支持异步调用和逐 token 流式输出。

痛点四：Python 绑定的局限性 1.0 整个框架只在 Python 中运行，但生产环境经常需要将链部署为独立的服务，或者在其他语言中调用。 解决方案：2.0 的 langserve 能把任何一个 LCEL 链一键转换成 REST API，配合 RemoteRunnable 可以跨进程、跨语言调用。未来还会原生支持 JavaScript，让同一套链定义在前端和后端都可运行。

结论：

如果 LangChain 1.0 是“让 LLM 应用能跑起来”的第一代引擎，那么 2.0 就是“让 LLM 应用能工程化、规模化生产”的重构。它用 LCEL 统一了抽象，用 Runnable 实现了真正的乐高式组装，用流式与异步适配了现代 Web，把 LLM 框架从“玩具拼装”带进了“工业制造”的轨道。

📋 4. LangChain 的 OutputParser 如何实现结构化输出？和 OpenAI 的 Function Calling 有什么区别？

一句话概括：

OutputParser 是在 LLM 的输出侧“套上一个格式化解析器”，强制把模型的自由文本转换成结构化的数据；而 Function Calling 是在 LLM 的输入侧“提前声明函数签名”，让模型自己输出一个结构化的函数调用请求。

1.1 OutputParser 的工作方式

LangChain 提供了多种解析器，核心思路是：你告诉模型“请按某种格式输出”，然后 Parser 把模型返回的原始字符串解析成你想要的数据结构。

Prompt (含格式指令) → LLM (输出符合指令的字符串) → OutputParser (解析为结构化对象)

三种典型实现：

查看内嵌表格

代码示例：用 PydanticOutputParser 提取人物信息

from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

# 1. 用 Pydantic 定义目标结构
class Person(BaseModel):
    name: str = Field(description="人物姓名")
    age: int = Field(description="年龄")
    city: str = Field(description="居住城市")

parser = PydanticOutputParser(pydantic_object=Person)

# 2. 在 Prompt 中注入格式指令
prompt = PromptTemplate(
    template="从以下文本中提取人物信息。\n{format_instructions}\n文本：{text}\n",
    input_variables=["text"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

# 3. 构建链并运行
chain = prompt | ChatOpenAI(model="gpt-4o") | parser
result = chain.invoke({"text": "张三今年30岁，住在上海。"})
print(result)  # Person(name='张三', age=30, city='上海')

parser.get_format_instructions() 会自动生成一段“请输出如下格式的 JSON”的指令，注入 prompt。模型看到这段指令后，输出符合 Pydantic schema 的 JSON，Parser 再将其转换为 Person 对象。

这就是 OutputParser 的本质：用 prompt 约束输出格式，再在代码层解析。

1.2 与 OpenAI Function Calling 的核心区别

查看内嵌表格

一个直观的对比：

# OutputParser 方式：模型输出文本，你解析
response = model.invoke("输出 JSON：姓名、年龄、城市")
# response 是 "{\"name\":\"张三\",\"age\":30,\"city\":\"上海\"}"
# 你需要自己 parse

# Function Calling 方式：模型直接给出函数调用
response = client.chat.completions.create(
    model="gpt-4o",
    tools=[{"type":"function","function":{"name":"extract_person","parameters":{...}}}]
)
# response.choices[0].message.tool_calls[0].function.arguments
# 直接是 {"name":"张三","age":30,"city":"上海"}，无需额外解析

选择建议：

• 如果你用的是不支持 Function Calling 的开源模型，OutputParser 是唯一选择。

• 如果你需要结构化输出，但不需要模型去“决策调用哪个函数”，用 Function Calling 中的 tool_choice: "required" 来做结构化输出，比 OutputParser 更可靠。

• 如果你想提取一个固定 schema 的数据，且用 Function Calling 的模型，那直接用 Function Calling。OutputParser 更适合那种“不依赖特定模型、需要框架无关性”的场景。

---

🤖 5. LangChain 中的 Agent 是如何实现 ReAct 模式的？执行循环的终止条件是什么？

5.1 ReAct 模式的内核

ReAct = Reasoning + Acting。Agent 在一个循环中交替进行“思考”和“行动”：

text

用户输入 → Agent 思考 → 决定调用工具 → 执行工具 → 观察结果 → 再思考 → ... → 最终回答

LangChain 中的实现核心是 AgentExecutor 和一个迭代循环。

它在背后做了几件事：

1. 组装 Prompt：把 System Prompt（含工具描述、ReAct 格式要求）、历史对话、用户输入一起塞给 LLM。

2. 解析 LLM 输出：从 LLM 的回复中提取出 “Action” 和 “Action Input”（要调用的工具和参数）。

3. 执行工具：根据解析结果，找到对应的 Tool 并调用它，拿到 Observation。

4. 追加 Observation：把工具执行的结果作为一条新的消息追加到对话历史中。

5. 循环：把更新后的对话历史再次发给 LLM，重复步骤 2-4，直到 LLM 输出的是 “Final Answer” 而不是 Action。

用代码拆解这个过程（简化版 AgentExecutor 内核）：

from langchain.agents import load_tools, initialize_agent, AgentType
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = load_tools(["serpapi", "llm-math"], llm=llm)
agent = initialize_agent(tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True)
agent.invoke({"input": "马斯克今年多少岁？他的年龄的平方根是多少？"})

当你调用 agent.invoke 时，内部发生的事情：

# 伪代码：AgentExecutor 循环的内部逻辑
def agent_execute(user_input):
    messages = [system_prompt_with_tools, user_input]
    for step in range(max_iterations):
        response = llm.invoke(messages)  # 1. LLM 思考

        if "Final Answer" in response:   # 2. 检查终止条件
            return extract_final_answer(response)

        action, action_input = parse_action(response)  # 3. 解析 Action
        observation = execute_tool(action, action_input)  # 4. 执行工具

        messages.append({"role": "assistant", "content": response})
        messages.append({"role": "user", "content": f"Observation: {observation}"})
    raise Exception("Agent 超过最大迭代次数仍未完成")

5.2 执行循环的终止条件

Agent 循环有两个终止条件，任何一个满足即停止：

① LLM 自主输出 Final Answer 当 LLM 认为“我已经收集了足够的信息，可以回答用户了”，它会在回复中包含 Final Answer: 标记。LangChain 的解析器检测到这个标记后，提取后面的内容作为最终回复。

Thought: 我知道马斯克年龄了，计算器可以算平方根。
Action: Calculator
Action Input: 52^(1/2)
Observation: 7.211...
Thought: 现在我可以给出最终答案了。
Final Answer: 马斯克今年52岁，52的平方根约为7.21。

② 达到最大迭代次数（Max Iterations） 你可以在 initialize_agent 时通过 max_iterations 参数设置上限（默认通常为 15）。如果 Agent 在这么多轮循环后还没输出 Final Answer，AgentExecutor 会抛出一个异常或强制返回最后一条 LLM 输出，防止无限循环消耗 Token。

agent = initialize_agent(tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
                         max_iterations=5,  # 限制最大轮次
                         early_stopping_method="force")  # 超时后强制返回最后一条消息

③ 额外的安全阀（可选）

• 早停策略：early_stopping_method="generate" 会在达到最大迭代次数时，把当前所有信息再喂给 LLM 一次，要求它强制给出 Final Answer。

• 超时控制：可以用 asyncio.timeout 或类似机制给整个 invoke 调用加一个硬时间上限，防止任务挂起。

• 工具异常处理：如果工具执行抛出异常，LangChain 会把异常信息作为 Observation 返回给 LLM，让 LLM 决定是否重试、换工具或提前结束。通常连续多次工具失败，LLM 会倾向于给出一个带有歉意和说明的 Final Answer。

总结 ReAct 循环：

它是 “试错-学习-再试”的自动化。LLM 是决策者，工具是执行者，Observation 是反馈信号。终止条件确保了流程不会永远卡住——要么 LLM 自己说“做完了”，要么系统强制叫停。这种设计让 Agent 在开放世界里既能自主探索，又不至于失控。

---

📚 6. 用 LangChain 2.0 LCEL 实现一个 RAG 问答链，如何设计链路？如何处理检索结果为空的情况？

RAG（检索增强生成）的核心流程是：先检索相关文档，再把文档和用户问题一起喂给 LLM，生成有依据的答案。

用 LCEL 来实现，我们可以把整个链路拆成几个独立的 Runnable 模块，然后用 | 管道符串联。

6.1 完整链路设计

6.2 LCEL 代码实现（带检索为空处理）

from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma

# 0. 准备向量存储和检索器
vectorstore = Chroma(embedding_function=OpenAIEmbeddings(), persist_directory="./chroma_db")
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# 1. 格式化文档的 Runnable
def format_docs(docs):
    """将检索到的文档列表拼接成一个字符串，并处理空结果"""
    if not docs:
        return "【注意：知识库中没有找到相关文档，请直接基于你的常识回答，但必须声明“未在知识库中找到相关信息”。】"
    return "\n\n".join(f"来源 {i+1}:\n{doc.page_content}" for i, doc in enumerate(docs))

format_docs_runnable = RunnableLambda(format_docs)

# 2. Prompt 模板（用 context 和 question 两个变量）
prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个智能问答助手。请根据以下提供的上下文回答用户问题。
要求：
1. 基于上下文给出准确答案。
2. 如果上下文中有明确答案，请引用来源编号。
3. 如果上下文中没有相关信息，请明确说明，并基于你的常识谨慎回答。
4. 保持回答简洁、专业。

上下文：
{context}"""),
    ("user", "{question}")
])

# 3. LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 4. 组装 LCEL 链
rag_chain = (
    {
        "context": retriever | format_docs_runnable,  # 检索 → 格式化
        "question": RunnablePassthrough()             # 直接传递用户问题
    }
    | prompt
    | llm
    | StrOutputParser()
)

# 5. 使用
answer = rag_chain.invoke("什么是向量检索的倒排索引？")
print(answer)

6.3 处理检索结果为空的完整策略

在上面的代码中，format_docs 函数已经实现了第一层处理：如果 docs 为空，返回一段特定的提示文本，强制 LLM 声明“未在知识库中找到相关信息”。

但在生产环境中，你可能需要更精细的控制：

查看内嵌表格

代码示例：用 RunnableBranch 实现条件路由

from langchain_core.runnables import RunnableBranch

# 判断检索结果是否为空的函数
def has_docs(input_dict):
    return len(input_dict["docs"]) > 0

# 有文档时的链
chain_with_docs = (
    {
        "context": RunnableLambda(lambda d: format_docs(d["docs"])),
        "question": RunnableLambda(lambda d: d["question"])
    }
    | prompt
    | llm
    | StrOutputParser()
)

# 无文档时的链：直接用 LLM 回答，但要求声明
prompt_no_docs = ChatPromptTemplate.from_template(
    "用户问题：{question}\n"
    "知识库中没有找到相关信息。请直接基于你的常识回答，"
    "但必须在回答开头声明“知识库未找到相关信息，以下为通用知识回答：”。"
)
chain_no_docs = (
    RunnableLambda(lambda d: {"question": d["question"]})
    | prompt_no_docs
    | llm
    | StrOutputParser()
)

# 用分支路由
branch_chain = RunnableBranch(
    (has_docs, chain_with_docs),
    chain_no_docs  # 默认分支（检索为空时）
)

# 先检索，再分支
full_chain = {
    "docs": retriever,
    "question": RunnablePassthrough()
} | branch_chain

answer = full_chain.invoke("什么是GPT-5？")

总结这个 RAG 链设计：

• 模块化：检索、格式化、Prompt、LLM、解析器，各司其职，都可单独测试和替换。

• 鲁棒性：通过 format_docs 注入引导语、RunnableBranch 条件路由、二次检索重试，确保即使知识库为空，系统也不会崩溃，而是给出合理回应。

• 可扩展：你可以轻松地在 retriever 之前加一个“问题改写”模块，或在 llm 之后加一个“答案验证”模块，都是 Runnable 的自由组合。

---

LangChain LCEL

LangChain LCEL 管道设计原理

🔗 7. 什么是 LCEL，| 操作符怎么用？

LCEL 是 LangChain Expression Language 的缩写。它不是另一种编程语言，而是用管道符 | 把处理单元串联起来的一种声明式语法。

想象一个物理实验：水从水龙头流出，经过滤芯，再流进杯子。每一个环节只做一件事，数据就像水一样单向流过各个部件。

在代码里，这个流程就是：

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

prompt = ChatPromptTemplate.from_template("给一家卖{product}的公司起三个名字")
model = ChatOpenAI()
parser = StrOutputParser()

# 这就是 LCEL 的 | 操作符
chain = prompt | model | parser

| 操作符到底做了什么？ 在 Python 里它被重载为 Runnable.or。简化的内部逻辑可以理解为：

class Runnable:
    def __or__(self, other):
        # 返回一个新的 RunnableSequence，调用时先运行 self，输出作为 other 的输入
        return RunnableSequence(self, other)

所以 prompt | model | parser 本质上创建了一个 RunnableSequence 对象。当你调用 chain.invoke({"product": "智能手表"}) 时：

1. prompt 把输入字典转换成 ChatPromptValue（一串消息对象）。

2. model 接收这些消息，输出一个 AIMessage。

3. parser 从 AIMessage 中提取出纯文本字符串。

| 不只用于串联，还能并联、分支。例如：

# 同时生成大纲和正文
chain = RunnableParallel(
    outline=prompt_outline | model | StrOutputParser(),
    article=prompt_article | model | StrOutputParser()
)

这里的 RunnableParallel 内部也是用 | 把多个子链并联起来。所以 | 就是 LCEL 的胶水——它让你能用搭积木的方式搭建复杂的数据流管道，而无需写任何胶水代码。

---

🧬 8. LCEL 管道设计原理是什么？它与旧版 LLMChain 的核心区别在哪里？

LCEL 的设计原理：一切皆 Runnable，统一接口，自由组合。

每个 LCEL 的组件（Prompt、LLM、Parser、Retriever、你自己写的函数）都遵循同一个 Runnable 协议：

• invoke(input) → output

• ainvoke(input) → output (异步)

• stream(input) → Iterator[output] (同步流)

• astream(input) → AsyncIterator[output] (异步流)

• batch(inputs) → list[output] (批量)

管道之所以能工作，是因为每个组件都有清晰的输入输出类型声明（通过 input_schema 和 output_schema）。| 操作符在组合时就会校验：前一个的输出模式是否和后一个的输入模式兼容。如果不兼容，会在链构建阶段直接报错，而不是等到运行时才挂。

核心区别：LCEL vs LLMChain

我用一张表格把它们的哲学差异说清楚：

查看内嵌表格

直观例子：实现“先总结、再翻译”

LLMChain 做法（需要嵌套，且调试黑盒）：

# 1.0 风格
summary_chain = LLMChain(llm=llm, prompt=summary_prompt)
translate_chain = LLMChain(llm=llm, prompt=translate_prompt)
# 你需要再写一个顺序链把它们包起来，而且没办法直接看中间结果

LCEL 做法（透明、可逐节测试）：

# 2.0 风格
summary_chain = summary_prompt | model | StrOutputParser()
translate_chain = translate_prompt | model | StrOutputParser()
full_chain = summary_chain | translate_chain  # 直接拼接
# 你可以单独测试 summary_chain 和 translate_chain，也可以从 full_chain 的 trace 里看到每一环的输入输出

用 LCEL 的本质转变：从“调用一个函数”到“定义一条数据流水线”。

这条流水线的节点可以随时被替换、重组，而且天然支持流式传输。这种声明式的力量，在下面流式返回的实现中会完全展现出来。

---

⚡ 9. 在 FastAPI 里如何用 LCEL 实现流式返回，RunnableParallel 怎么做多路并发？

流式返回是 LLM 应用体验的分水岭：用户不想等 10 秒看完整答案，而是希望看到字一个个蹦出来。LCEL 让这件事极度简单。

9.1 FastAPI + LCEL 流式返回

原理：调用 LCEL 链的 astream 方法，会返回一个 AsyncIterator，每次 yield 一个增量。FastAPI 的 StreamingResponse 可以直接消费这个异步迭代器，并封装成 Server-Sent Events（SSE）格式。

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
import asyncio

app = FastAPI()

# 构建一个简单的 LCEL 链
prompt = ChatPromptTemplate.from_template("请用中文回答：{question}")
model = ChatOpenAI(model="gpt-4o", streaming=True)  # streaming=True 很重要
chain = prompt | model | StrOutputParser()

async def generate(question: str):
    """异步生成器，逐块产出 LLM 的回答"""
    async for chunk in chain.astream({"question": question}):
        # chunk 是每次新增的文本片段，如 "今天"、"天气"、"很好"
        yield f"data: {chunk}\n\n"
        await asyncio.sleep(0)  # 让出控制权，确保客户端及时收到
    yield "data: [DONE]\n\n"

@app.get("/stream")
async def stream_endpoint(q: str = "什么是LCEL？"):
    return StreamingResponse(
        generate(q),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}
    )

关键点：

• model 必须开启 streaming=True，否则底层不会流式输出。

• astream 返回的是 AsyncIterator[str]，每个 str 是一个或多个 token 的增量。

• StreamingResponse 消费异步生成器，SSE 格式让前端可以用 EventSource 轻松对接。

9.2 RunnableParallel 实现多路并发

当你的任务需要同时从不同角度处理数据时——比如用户问“分析新能源汽车市场”，你既要生成一份市场报告大纲，又要生成一份竞争对手清单——RunnableParallel 能让这两件事并行执行，而不是串行等待。

原理：RunnableParallel 接收一个字典，键是分支名，值是一条 LCEL 子链。当调用 invoke 或 astream 时，所有分支会同时执行，最后合并成一个字典输出。

from langchain_core.runnables import RunnableParallel, RunnablePassthrough

# 大纲生成链
outline_prompt = ChatPromptTemplate.from_template("为关于{topic}的市场报告生成大纲")
outline_chain = outline_prompt | model | StrOutputParser()

# 竞品清单链
competitor_prompt = ChatPromptTemplate.from_template("列出{topic}领域的5家主要竞争对手")
competitor_chain = competitor_prompt | model | StrOutputParser()

# 并行链
parallel_chain = RunnableParallel(
    outline=outline_chain,
    competitors=competitor_chain
)

# 调用：两个分支同时执行，最终返回字典
result = parallel_chain.invoke({"topic": "新能源汽车"})
print(result["outline"])      # 市场报告大纲
print(result["competitors"])  # 竞争对手列表

如果要在 FastAPI 中流式返回并行结果怎么办？

我们可以并发地跑两条流，然后把它们的 token 合并成一个 SSE 流，用前缀区分来源。

@app.get("/stream/parallel")
async def stream_parallel(topic: str = "新能源汽车"):
    async def event_generator():
        # 并发启动两个流
        async def stream_branch(name, chain):
            async for chunk in chain.astream({"topic": topic}):
                yield f"data: [{name}] {chunk}\n\n"
            yield f"data: [{name}] DONE\n\n"

        # 用 asyncio.gather 并发执行两个生成器，需要特殊处理
        # 为了把它们交织发送，可以用队列
        queue = asyncio.Queue()

        async def producer(name, chain):
            async for chunk in chain.astream({"topic": topic}):
                await queue.put(f"data: [{name}] {chunk}\n\n")
            await queue.put(f"data: [{name}] DONE\n\n")

        # 启动两个生产者
        tasks = [
            asyncio.create_task(producer("outline", outline_chain)),
            asyncio.create_task(producer("competitors", competitor_chain))
        ]

        # 消费者：从队列取消息发送，直到两个生产者都完成并且队列为空
        finished = 0
        while finished < 2:
            msg = await queue.get()
            yield msg
            if "DONE" in msg:
                finished += 1

        await asyncio.gather(*tasks)
        yield "data: [ALL_DONE]\n\n"

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

这段代码的核心：

• asyncio.Queue 让两个并发的生成器安全地把数据塞进同一个队列。

• 消费者从队列取出数据，即时发送给客户端，实现了真正的多路并发流式响应。

前端只要根据 [outline]、[competitors] 前缀，就能把同时到达的消息渲染到页面的不同区域，实现类似“分屏直播”的效果。

LangChain 的自定义 Tool 定义与 Pydantic Schema 集成

⚠️ 10. LangChain 中 @tool 装饰器的最关键注意事项

@tool 装饰器是 LangChain 中最快捷的定义工具的方式：只需给一个 Python 函数加上 @tool，框架就会自动提取函数名、文档字符串（docstring）和参数类型，生成一个 Tool 对象。

最关键的一个注意事项是：函数的 docstring 和参数类型注解会直接决定 LLM 调用工具的准确性，绝对不能忽略或写得太随意。

工具定义 → LLM 阅读工具描述 → 决定是否调用 → 生成参数

LLM 看不到你的函数体代码，它唯一能看到的就是：

• 函数名（get_weather）

• docstring（工具描述）

• 参数名和类型注解（city: str）

• 参数默认值（如果有）

所以，@tool 的核心原则是：把 docstring 当作给 LLM 的说明书来写，把类型注解当作参数契约来定义。

错误示例 vs 正确示例：

# ❌ 差：LLM 不知道参数格式，容易调用失败
@tool
def get_weather(city):
    """获取天气"""
    return f"{city}: 晴"

# ✅ 好：明确的参数类型、描述、格式约束
@tool
def get_weather(city: str) -> str:
    """获取指定城市的当前天气，包括温度、湿度和天气状况。

    参数:
        city: 城市名称，必须是中文拼音，例如 Beijing, Shanghai
    """
    return f"{city}: 25°C, 晴"

为什么这一点最关键？ 因为参数类型注解不仅影响 Python 类型检查，更会被 LangChain 转换为 JSON Schema 并呈现给 LLM。如果你不写类型注解，LangChain 会默认参数类型为 string，但你永远无法约束格式；如果你把 date: str 写成 date，LLM 可能会填入 "今天" 而不是 "2026-07-02"。

另外三个辅助注意事项：

• 返回类型注解（-> str）虽然不直接影响 LLM，但能帮助 LangChain 生成更准确的输出描述。

• 如果参数有有限选项，最好用 Literal 类型，这会生成 enum 约束，让 LLM 只能从候选中选择。

• 启用 return_direct=True 可以让工具结果直接返回给用户，跳过 LLM 的二次总结，适用于查天气、查股价等“结果即答案”的场景。

from typing import Literal
from langchain.tools import tool

@tool
def set_volume(level: Literal["low", "medium", "high"]) -> str:
    """设置音量级别。level 只能是 low, medium, high 其中之一。"""
    return f"音量已设为 {level}"

一句话总结： 你写 @tool 的时候，要时刻想“LLM 只读得到我的 docstring 和参数名”，所以把这两个写到极致清晰。

---

🛠️ 11. LangChain 定义自定义 Tool 的三种方式及 Pydantic Schema 场景

LangChain 提供了三种定义 Tool 的方式，灵活度从低到高，适用场景也各不相同。

方式一：@tool 装饰器（最简便）

适用场景：函数逻辑简单、参数自动推导已足够、快速原型开发。

from langchain.tools import tool

@tool
def multiply(a: int, b: int) -> int:
    """两个整数相乘"""
    return a * b

内部 LangChain 会用 infer_schema 从函数签名自动生成 JSON Schema。但如果你需要更复杂的嵌套结构或字段描述，这种方式就不够用了。

方式二：StructuredTool.from_function + Pydantic Schema（最推荐生产使用）

适用场景：需要精细控制参数描述、嵌套对象、自定义校验逻辑。

这是最推荐的生产级方式，因为它允许你手动定义一个 Pydantic 模型来精确描述参数，而不依赖自动推断。StructuredTool.from_function 接收这个 Pydantic 模型作为 args_schema。

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

# 1. 定义 Pydantic 模型（即工具的输入 schema）
class WeatherInput(BaseModel):
    city: str = Field(description="城市名称，如 Beijing, Shanghai")
    date: str = Field(description="日期，格式 YYYY-MM-DD，如 2026-07-02")

# 2. 定义执行函数（参数名和 Pydantic 字段一致）
def get_weather(city: str, date: str) -> str:
    return f"{city} 在 {date} 的天气：晴，25°C"

# 3. 用 from_function 创建工具
weather_tool = StructuredTool.from_function(
    func=get_weather,
    name="get_weather",
    description="获取指定城市在指定日期的天气",
    args_schema=WeatherInput,
    return_direct=True  # 结果直接返回用户
)

为什么这样更好？ Field(description=...) 里的文本会原封不动地出现在 LLM 看到的 JSON Schema 中，成为 LLM 填写参数的指南。你可以自由添加任何 Pydantic 支持的校验器，如 gt（大于）、regex 等，这些都会反映在 Schema 的约束中，让 LLM 更不容易填错。

方式三：继承 BaseTool（最灵活）

适用场景：工具需要管理内部状态（如数据库连接池）、需要异步初始化、或需要在执行前后做钩子操作。

from langchain.tools import BaseTool
from pydantic import BaseModel, Field

class DatabaseInput(BaseModel):
    query: str = Field(description="SQL 查询语句，只支持 SELECT")

class DatabaseTool(BaseTool):
    name: str = "query_database"
    description: str = "执行 SQL 查询，返回结果列表"
    args_schema: type[BaseModel] = DatabaseInput

    # 可以在这里初始化连接池
    def _run(self, query: str) -> str:
        # 同步执行逻辑
        return execute_query(query)

    # 如果需要异步
    async def _arun(self, query: str) -> str:
        return await async_execute_query(query)

场景：你有一个需要保持长连接的数据库查询工具，或者你想在每次调用前后自动记录审计日志，继承 BaseTool 让你完全控制执行过程。

选型建议：

查看内嵌表格

---

🎯 12. Agent 频繁调用工具时参数不合法，如何从 Tool 定义层面减少这类错误？

Agent 调用工具参数不合法的常见原因有：

• 参数描述模糊，LLM 填入了非法值（如负数、错误日期格式）。

• 枚举值未约束，LLM 自己编造了不存在的选项。

• 缺乏示例，LLM 不理解边界情况。

从 Tool 定义层面，有五项可以立即落地的优化策略：

策略一：用 Pydantic 的约束类型明确边界

不要在 docstring 里口头说“必须是正整数”，而要用 Pydantic 的 Field(gt=0)、min_length、regex 等硬约束。

from pydantic import BaseModel, Field, conint

class OrderInput(BaseModel):
    quantity: conint(gt=0, le=100) = Field(description="购买数量，必须在 1 到 100 之间")
    email: str = Field(description="用户邮箱", pattern=r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$")

这些约束会转换成 JSON Schema 中的 minimum、maximum、pattern 等字段，LLM 看到后会显著降低填错概率。

策略二：用 Literal 限制枚举选项

如果参数只有几个固定值，必须用 Literal，不要写成 str 然后在描述里列选项。

from typing import Literal

class SortInput(BaseModel):
    sort_by: Literal["price", "rating", "sales"] = Field(description="排序字段")
    order: Literal["asc", "desc"] = Field(description="排序方向")

LLM 看到 enum: ["price", "rating", "sales"] 后，几乎不可能再编造一个 "popularity" 出来。

策略三：在 description 中嵌入具体示例

大模型对示例的遵从度远高于对规则的遵从度。在 Field(description=...) 里直接写一两个正确值示例。

class DateInput(BaseModel):
    date: str = Field(description="查询日期，格式 YYYY-MM-DD，例如 2026-07-02")

不要只写“格式为 YYYY-MM-DD”，加上一个具体日期，LLM 会照着模仿。

策略四：在描述中定义“失败后果”，引导 LLM 谨慎填写

人会对 LLM 说“如果填错会返回错误”，LLM 也会因此更仔细。

class TransferInput(BaseModel):
    amount: float = Field(description="转账金额，必须大于 0 且不超过账户余额，否则转账失败")

这种“自然后果”式的描述，比硬规则更能激活模型的遵从意识。

策略五：启用 handle_parsing_errors 并返回清晰错误信息

这不是 Tool 定义本身，而是 Agent 配置层面的防御。当参数解析失败时，把具体错误反馈给 LLM，让它自我修正。

agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    handle_parsing_errors=True,  # 自动捕获解析错误
    max_iterations=5
)

如果工具参数不合法导致执行失败，你可以把错误信息包装成 Observation 返回：

def robust_tool(**kwargs):
    try:
        validate_params(kwargs)
        return actual_execute(**kwargs)
    except ValueError as e:
        return f"参数错误: {e}。请检查并重试。"

LLM 看到这个错误后，通常能在下一轮给出正确参数。

一个综合了以上策略的生产级工具定义：

from pydantic import BaseModel, Field
from typing import Literal
from langchain.tools import StructuredTool

class FlightSearchInput(BaseModel):
    origin: str = Field(description="出发城市，IATA 三字码，如 PEK")
    destination: str = Field(description="到达城市，IATA 三字码，如 SHA")
    date: str = Field(description="出发日期，格式 YYYY-MM-DD，例如 2026-07-15")
    cabin: Literal["economy", "business", "first"] = Field(description="舱位等级")
    passengers: int = Field(description="乘客数量", ge=1, le=9)

def search_flights(origin, destination, date, cabin, passengers):
    # 实际查询逻辑
    pass

flight_tool = StructuredTool.from_function(
    func=search_flights,
    name="search_flights",
    description="搜索航班信息，返回航班号、起降时间、价格",
    args_schema=FlightSearchInput,
    handle_tool_error=True  # 如果函数内抛异常，自动转为错误消息给 LLM
)

总结：

工具定义是 Agent 和真实世界之间的“闸门”。你在 Pydantic Schema 中花的每一分钟，都会在运行时节省无数次失败的函数调用。记住，LLM 只看得到 Schema 和描述，所以把这两样写到“即便是一个认真的实习生”也绝不会填错的程度，你的 Agent 才能稳健地工作在生产环境里。

LangChain 深度进阶

[LCEL、Memory 机制、Retriever 类型、生产优化]

⚡ 13. LangChain 的 LCEL 是什么？有什么优势？

LCEL（LangChain Expression Language） 是 LangChain 2.0 的核心设计范式，一种用管道符 | 把处理单元串联成数据流管道的声明式语言。

每个组件都是 Runnable 对象，拥有统一的接口：invoke、ainvoke、stream、astream、batch。| 操作符把它们首尾相连，前一个的输出自动成为后一个的输入。

代码示例：一个简单的 LCEL 链

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

prompt = ChatPromptTemplate.from_template("给一家卖{product}的公司起三个名字")
model = ChatOpenAI(model="gpt-4o", temperature=0.8)
chain = prompt | model | StrOutputParser()

names = chain.invoke({"product": "智能猫砂盆"})
print(names)

五大核心优势：

1. 声明式组合 不需要写胶水代码，管道符直接表达数据流向。一个链可以轻易拆成几段独立测试，也可以像乐高一样重新拼装。

2. 统一的 Runnable 接口 所有组件都支持同步/异步、批量、流式调用。你想从 invoke 切换到 astream 做流式输出，只需把调用方法换一下，链本身无需任何修改。

3. 自动并行与优化 RunnableParallel 能自动识别无依赖的分支，并发执行它们。RunnablePassthrough 让你灵活透传原始输入，配合字典拆分实现复杂的多路处理。

from langchain_core.runnables import RunnableParallel
# 同时生成大纲和正文
parallel_chain = RunnableParallel(
    outline=outline_chain,
    article=article_chain
)
result = parallel_chain.invoke({"topic": "AI"})
# result 是 {"outline": "...", "article": "..."}

1. 原生流式支持 任何一个 LCEL 链，直接 .stream() 就能逐 token 输出，大幅提升用户体验。

2. 可观测性 每一步的输入输出都被透明追踪，配合 LangSmith 可以一眼看到数据在哪一步变质，调试效率成倍提升。

一句话概括：LCEL 让你从“调用函数”的思维，切换到“定义数据流水线”的思维。 这条流水线上的每个节点都可以独立替换、测试、复用，这是 LangChain 2.0 工程化能力飞跃的根基。

---

🧠 14. LangChain 的 Memory 机制有哪几种？各自适用什么场景？

Memory 是 LangChain 中负责跨轮次保持状态的模块。它的本质是把历史对话或摘要注入到 LLM 的上下文窗口，让无状态的模型“记住”之前发生的事情。

用户输入 → [Memory 注入历史] → 完整 Prompt → LLM → 回复 → [Memory 更新]

六种常用 Memory 及适用场景：

① ConversationBufferMemory（全量缓存）

原理：原封不动地存储所有对话消息，每次请求时把完整历史注入 prompt。

适用场景：短对话、原型开发、token 预算充裕时。

局限：对话一长就会撑爆上下文窗口。

from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory(return_messages=True)

② ConversationBufferWindowMemory（滑动窗口）

原理：只保留最近 K 轮对话，旧消息自动丢弃。 适用场景：中等长度对话，关心近期上下文，不想为远古历史浪费 token。 配置：k=5 表示保留最近 5 轮。

from langchain.memory import ConversationBufferWindowMemory
memory = ConversationBufferWindowMemory(k=5, return_messages=True)

③ ConversationSummaryMemory（摘要记忆）

原理：用 LLM 将对话历史压缩成一段摘要，只存摘要不存原文。

适用场景：长对话、用户和助手交互非常多轮，需要保留全局脉络但不想逐字存储。

代价：每次都需要额外调用 LLM 做摘要，增加延迟和费用。

from langchain.memory import ConversationSummaryMemory
memory = ConversationSummaryMemory(llm=llm)

④ ConversationSummaryBufferMemory（混合记忆）

原理：结合滑动窗口和摘要。最近 K 轮保留原文，更早的历史用摘要压缩。 适用场景：生产环境长对话的最佳平衡方案——既有近期的细节，又有远期的脉络，同时严格控制 token 消耗。 配置：max_token_limit 控制最大 token 数。

from langchain.memory import ConversationSummaryBufferMemory
memory = ConversationSummaryBufferMemory(llm=llm, max_token_limit=500)

⑤ ConversationTokenBufferMemory（Token 限额缓存）

原理：和 BufferWindow 类似，但按 token 数量而非轮次来截断。

适用场景：模型有严格 token 上限，需要精准控制 prompt 长度。

from langchain.memory import ConversationTokenBufferMemory
memory = ConversationTokenBufferMemory(llm=llm, max_token_limit=1000)

⑥ VectorStoreRetrieverMemory（向量记忆）

原理：把对话片段存入向量数据库，每次根据当前问题检索最相关的历史记忆片段。

适用场景：长期、跨会话的记忆。Agent 需要从很久以前的对话中回忆起用户偏好、历史决策等。

from langchain.memory import VectorStoreRetrieverMemory
memory = VectorStoreRetrieverMemory(retriever=vectorstore.as_retriever())

选型速查表：

查看内嵌表格

核心设计思想： 上下文窗口是 LLM 最稀缺的资源。Memory 的本质就是用工程手段在有限窗口里装下最有价值的历史信息。选择哪一种，取决于你更看重“近期细节”还是“全局脉络”。

---

🔍 15. LangChain 的 Retriever 有哪些类型？如何实现混合检索？

Retriever 是 LangChain 中负责从外部知识库检索相关文档的组件，是 RAG 系统的核心引擎。

用户问题 → Retriever → 相关文档列表 → 拼入 Prompt → LLM 生成答案

五种常见 Retriever 类型：

① VectorStoreRetriever（向量检索）

原理：将问题和文档都用 Embedding 模型向量化，通过余弦相似度检索 top-K 最相关文档。

适用场景：语义匹配，擅长找“意思相近”的内容。

局限：对精确关键词匹配（如专有名词、编号）较弱。

from langchain_chroma import Chroma
retriever = Chroma(embedding_function=embeddings).as_retriever(search_kwargs={"k": 4})

② BM25Retriever（关键词检索）

原理：基于 TF-IDF 的稀疏向量检索，擅长精确关键词匹配。

适用场景：搜索特定术语、ID、人名等。

局限：不理解语义，同义词搜索不到。

from langchain_community.retrievers import BM25Retriever
retriever = BM25Retriever.from_documents(docs, k=4)

③ MultiQueryRetriever（多查询检索）

原理：用 LLM 把用户问题改写成多个不同角度的查询，分别检索后合并结果。

适用场景：用户问题太口语化或太宽泛，需要从多角度覆盖。

优势：提升召回率，减少漏检。

from langchain.retrievers import MultiQueryRetriever
retriever = MultiQueryRetriever.from_llm(retriever=base_retriever, llm=llm)

④ ContextualCompressionRetriever（上下文压缩检索）

原理：先粗检索一堆文档，再用 LLM 从中提取与问题最相关的片段，过滤掉无关内容。

适用场景：检索到的文档很长，但只有一小段有用。

优势：提高信息密度，减少 prompt 噪音。

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
compressor = LLMChainExtractor.from_llm(llm)
retriever = ContextualCompressionRetriever(base_retriever=base, compressor=compressor)

⑤ EnsembleRetriever（集成检索）—— 实现混合检索的关键

原理：组合多个不同类型的 Retriever，用 Reciprocal Rank Fusion (RRF) 等算法融合它们的排序结果。

适用场景：你需要同时利用语义相似度和精确关键词匹配的混合检索。

实现混合检索（向量 + BM25）的完整代码：

from langchain_chroma import Chroma
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever
from langchain_openai import OpenAIEmbeddings

# 1. 准备文档（示例）
docs = [
    "报告编号 ABC-1234：2025年度销售总结",
    "苹果公司发布了新款 iPhone，售价 799 美元",
    "香蕉富含钾元素，有益健康",
    "如何制作苹果派：食谱和步骤"
]

# 2. 构建向量检索器（语义匹配）
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_texts(docs, embeddings)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 3. 构建 BM25 检索器（关键词匹配）
bm25_retriever = BM25Retriever.from_texts(docs, k=3)

# 4. 集成检索器：权重向量检索 60%，BM25 40%
ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.6, 0.4]  # 权重分配
)

# 5. 测试
results = ensemble_retriever.invoke("苹果产品的价格")
for doc in results:
    print(doc.page_content)

混合检索的权重调优：

• 如果你的用户查询偏自然语言描述，给向量检索更高权重（0.7～0.8）。

• 如果涉及大量精确匹配（如文档编号、错误码），给 BM25 更高权重。

• 可以用离线评估集（query + 人工标注的相关文档）来网格搜索最优权重。

收束：

Retriever 是 RAG 系统的“第一公里”，决定了 LLM 能看见什么信息。单一检索方式总有死角，向量检索管语义，BM25 管关键词，多查询管覆盖面，压缩器管密度，集成检索管融合。把这五种能力组合好，你的知识库问答系统才能真正做到“又全又准”。

---