AI Agent 工程实践教程 · 第 07 章

从 Prompt、Model、OutputParser 到 Chain，掌握 LangChain 的工程化组合方式。

返回系列目录

第七天：LangChain 框架实战

前面几天，我们已经把学生端 AI 问答的关键能力做出来了：

1. Python 服务可以调用大模型。
2. Java 负责学生鉴权、会话保存、历史消息查询。
3. Python 可以接收历史消息并组装 messages。
4. 知识库问答已经接入 LlamaIndex 和 Milvus。
5. 工具调用已经能把“查课表”这类真实业务能力交给 AI 使用。

所以第七天不再重复“怎么调用大模型”。

这一节课要解决的是另一个问题：

当一个 AI 应用里同时有提示词、模型、历史消息、知识库、工具、流式输出时，Python 代码应该怎么组织，才不会越来越乱？

这就是 LangChain 要解决的问题。

一句话：

直接调大模型，是调用能力；LangChain，是组织能力。

LangChain 不是大模型。

LangChain 也不是知识库。

LangChain 更像是 AI 应用里的“流程编排层”：

输入问题
  ↓
准备上下文
  ↓
套入 Prompt 模板
  ↓
调用模型
  ↓
解析模型输出
  ↓
必要时接入知识库或工具
  ↓
返回结果

1. 为什么直接调大模型还不够

在最简单的 AI 应用里，Python 直接调大模型就可以。

比如：

用户问题 → 调用模型 → 返回回答

这个流程很短，手写代码完全没问题。

但是我们的学生端 AI 问答不是一个简单问答接口。

它现在已经有很多步骤：

学生提问
  ↓
Java 校验学生身份
  ↓
Java 查询历史消息
  ↓
Python 判断是否需要知识库
  ↓
需要时用 LlamaIndex 检索资料
  ↓
Python 获取当前生效的系统提示词
  ↓
Python 组装 messages
  ↓
Python 判断是否开放工具
  ↓
需要时让模型选择工具
  ↓
Python 执行业务工具
  ↓
模型根据工具结果生成最终回答
  ↓
流式返回给 Java 和前端

如果所有步骤都写成普通函数互相调用，前期看起来没问题，但后面会遇到几个麻烦：

1. 每加一个步骤，就要改一段流程代码。
2. Prompt、模型、解析器、检索函数、工具函数混在一起。
3. 想复用某个步骤比较麻烦。
4. 想单独调试某个环节不方便。
5. 想把普通回答、RAG 回答、工具回答组织成统一风格不容易。

LangChain 的价值不是让模型变聪明。

它的价值是：

让 AI 应用里的每个步骤都变成可以组合的组件。

也可以这样理解：

不用 LangChain：
一堆 Python 代码手动串流程。

使用 LangChain：
把 Prompt、Model、Parser、函数、检索器、工具都变成 Runnable，再组合成 Chain。

2. LangChain 最核心的思想

学习 LangChain 时，不需要一开始就记住所有概念。

先记住一条最重要的链：

chain = prompt | model | parser

这条链是 LangChain 入门最关键的代码。

它可以拆成三步理解：

Prompt
  负责把输入变量变成模型需要的 messages。

Model
  负责真正调用大模型。

Parser
  负责把模型返回结果转换成程序更容易使用的格式。

例如学生输入：

什么是向量数据库？

LangChain 的执行过程大概是：

{"question": "什么是向量数据库？"}
  ↓
ChatPromptTemplate 生成 messages
  ↓
ChatOpenAI / 其他模型对象调用大模型
  ↓
StrOutputParser 把 AIMessage 转成字符串
  ↓
"向量数据库是一种用来存储和检索向量数据的数据库..."

这里的 | 很重要。

它不是字符串拼接。

它表示：

把上一步的输出，交给下一步继续处理。

这就是 LangChain 的链式写法，也叫 LCEL。

LCEL 可以先理解成：

LangChain 里用来组合流程的一种写法。

3. 课程环境和依赖

如果要在 Python 项目里使用 LangChain，通常需要安装：

uv add langchain langchain-core langchain-openai

这里说明一下：

1. langchain：LangChain 主包。
2. langchain-core：核心抽象，比如 Prompt、Runnable、Parser、Message。
3. langchain-openai：OpenAI 兼容模型接入包。

我们的 Python 服务目前使用的是百炼 OpenAI 兼容接口。

也就是说，虽然模型不是 OpenAI 官方模型，但接口形式兼容 OpenAI Chat Completions，所以可以用 langchain-openai 里的 ChatOpenAI 来接。

后面代码会用到这些配置：

BAILIAN_API_KEY
BAILIAN_BASE_URL
BAILIAN_MODEL

这些配置已经在项目的 Settings 里有类似字段：

bailian_api_key
bailian_base_url
bailian_model

下面先用一个独立 demo 练习 LangChain API。

真实项目里不要把这些配置直接写在代码里，而是从 Settings 读取。

4. 创建模型对象

LangChain 里，模型一般会先被封装成一个对象。

对我们项目来说，可以这样创建一个兼容百炼接口的模型对象：

from langchain_openai import ChatOpenAI


model = ChatOpenAI(
    model="qwen-plus",
    api_key="你的 BAILIAN_API_KEY",
    base_url="你的 BAILIAN_BASE_URL",
    temperature=0.3,
    max_tokens=1000,
)

这几个参数要讲清楚：

在真实项目里，不要把 API Key 写死在代码里。

更合理的写法是：

from langchain_openai import ChatOpenAI
from yanque_ai.core.config import settings


model = ChatOpenAI(
    model=settings.bailian_model,
    api_key=your_placeholder
    base_url=settings.bailian_base_url,
    temperature=0.3,
    max_tokens=1000,
)

这里要理解：

LangChain 的模型对象，本质上是对底层模型 API 的封装。

我们不再直接写：

client.chat.completions.create(...)

而是使用统一的：

model.invoke(...)
model.stream(...)
model.batch(...)

5. invoke：普通模型调用

invoke 表示执行一次调用，等模型完整回答生成完，再返回结果。

response = model.invoke("什么是 LangChain？")

print(response.content)

这里返回的 response 通常是一个 AIMessage。

它不是普通字符串。

如果要拿到文本，要用：

response.content

AIMessage 里常见的字段有这些：

刚开始写业务代码时，最常用的就是：

print(response.content)

后面使用 StrOutputParser()，就是为了把 AIMessage 里的 content 直接取出来，让链路最后返回普通字符串。

也可以传入 messages：

from langchain_core.messages import SystemMessage, HumanMessage


messages = [
    SystemMessage(content="你是一个耐心、专业的 AI 课程助教。"),
    HumanMessage(content="请解释一下什么是 LangChain。"),
]

response = model.invoke(messages)

print(response.content)

这里和前面课程里的 messages 是同一个思想。

只是 LangChain 把字典形式的消息封装成了对象。

所以：

SystemMessage(content="你是 AI 课程助教")

大致等价于：

{
  "role": "system",
  "content": "你是 AI 课程助教"
}

6. stream：流式输出

聊天页面通常不希望等完整答案生成完再显示。

如果回答很长，用户会觉得页面卡住了。

所以前端聊天经常使用流式输出。

LangChain 模型对象可以这样流式调用：

for chunk in model.stream("请解释一下 RAG 的工作流程"):
    print(chunk.content, end="", flush=True)

有些版本或模型返回的 chunk 上也可以使用 text() 或类似辅助方法。

刚开始先记住最朴素的方式：

chunk.content

流式输出要注意三件事：

1. 每个 chunk 只是一小段内容，不是完整回答。
2. 前端要把多个 chunk 拼起来显示。
3. 如果要保存完整回答，需要后端自己累积完整文本。

这和我们项目现在的流式处理是一样的思想。

项目里 Python 逐段返回文本，Java 和前端逐段接收。

最终 Java 还要保存完整 AI 回复。

所以这里要记住：

stream 解决的是用户体验问题。
不是模型答案变了，而是返回方式变了。

7. batch：批量调用

batch 用来一次处理多个输入。

例如：

questions = [
    "什么是 RAG？",
    "什么是向量数据库？",
    "什么是 Prompt？",
]

responses = model.batch(questions)

for response in responses:
    print(response.content)

batch 不太适合普通聊天，因为聊天是一问一答。

但它适合这些场景：

1. 批量生成题目解析。
2. 批量给学生问题分类。
3. 批量把文档标题改写成更清楚的标题。
4. 批量对知识库 chunk 做摘要。

这里要记住：

invoke：一个输入，一次完整回答。
stream：一个输入，逐段返回。
batch：多个输入，批量处理。

8. ChatPromptTemplate：提示词模板

前面第五天已经讲过提示词工程。

真实项目里，不应该到处写字符串拼接。

例如不要这样写：

prompt = "你是 AI 问答助手，请回答：" + question

这种写法的问题是：

1. 系统规则不清楚。
2. 用户变量容易和规则混在一起。
3. 后面要加知识库资料、历史消息时会越来越乱。
4. 不方便复用和测试。

LangChain 提供了 ChatPromptTemplate。

它专门用来管理聊天模型的提示词模板。

from langchain_core.prompts import ChatPromptTemplate


prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个耐心、专业的 AI 课程助教。"),
    ("human", "请用学生能听懂的话解释：{question}"),
])

这段代码里有两个重点：

1. system 放固定规则。
2. {question} 是运行时传入的变量。

可以单独运行 prompt：

messages = prompt.invoke({
    "question": "什么是向量数据库？"
})

print(messages)

这一步不会调用大模型。

它只是把变量填入模板，生成模型需要的 messages。

可以这样理解：

ChatPromptTemplate 的职责：
把输入变量变成 messages。

8.1 另外两个相关模板

如果在编辑器里查看 ChatPromptTemplate，可能还会看到两个相关类：

FewShotChatMessagePromptTemplate
StructuredPrompt

这两个不需要一开始就深入掌握，先知道它们分别解决什么问题即可。

FewShotChatMessagePromptTemplate：把示例问答放进 Prompt

FewShot 的意思是“给模型几个示例”。

比如希望模型用固定风格回答，就可以先给它几组示例：

from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate


examples = [
    {
        "question": "什么是 RAG？",
        "answer": "RAG 是先检索资料，再让模型基于资料回答。",
    },
    {
        "question": "什么是 Tool？",
        "answer": "Tool 是可以交给 AI 调用的业务能力。",
    },
]

example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{question}"),
    ("ai", "{answer}"),
])

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

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个 AI 课程助教，请模仿示例的简洁风格回答。"),
    few_shot_prompt,
    ("human", "{question}"),
])

它最终生成的 messages 里，会先放入示例问答，再放入这次真正的问题。

适合这种场景：

我不只是想告诉模型规则，
还想给它几个回答样例，让它模仿样例的格式和风格。

课堂演示代码：

test/langchain_demo/16_few_shot_prompt_demo.py

StructuredPrompt：提示模型按结构输出

StructuredPrompt 更偏高级一点。

它的目标是：

不只让模型回答一段文字，
而是希望模型按指定结构返回内容。

例如希望模型解释一个课程概念时，固定包含：

name：概念名称
explanation：解释
example：例子

可以先定义一个结构：

from pydantic import BaseModel, Field
from langchain_core.prompts.structured import StructuredPrompt


class CourseConcept(BaseModel):
    name: str = Field(description="概念名称")
    explanation: str = Field(description="适合初学者理解的解释")
    example: str = Field(description="一个简单例子")


prompt = StructuredPrompt.from_messages_and_schema(
    [
        ("system", "你是一个 AI 课程助教。"),
        ("human", "请解释这个概念：{concept}"),
    ],
    schema=CourseConcept,
)

这一块会牵扯结构化输出、Pydantic 和模型兼容性。

所以第七天只需要先知道：

普通聊天提示词：优先用 ChatPromptTemplate。
需要给示例：可以了解 FewShotChatMessagePromptTemplate。
需要结构化输出：可以了解 StructuredPrompt。

课堂演示代码：

test/langchain_demo/17_structured_prompt_demo.py

9. 把 Prompt 和 Model 串起来

有了 prompt 和 model 以后，就可以组合成一条链。

chain = prompt | model

调用：

response = chain.invoke({
    "question": "什么是 RAG？"
})

print(response.content)

执行顺序是：

输入 {"question": "什么是 RAG？"}
  ↓
prompt 填充模板，生成 messages
  ↓
model 调用大模型，生成 AIMessage

这个写法比手动分开写更清楚：

messages = prompt.invoke({"question": "什么是 RAG？"})
response = model.invoke(messages)

两种写法本质一样。

区别是：

手动写法适合理解过程。
链式写法适合组织流程。

10. StrOutputParser：把输出变成字符串

模型返回的通常是 AIMessage。

但是很多业务代码只需要字符串。

例如前端显示、接口返回、日志记录，一般都只关心文本内容。

这时可以使用 StrOutputParser：

from langchain_core.output_parsers import StrOutputParser


parser = StrOutputParser()

组合成完整链路：

chain = prompt | model | StrOutputParser()

answer = chain.invoke({
    "question": "什么是 RAG？"
})

print(answer)

这时 answer 就是字符串。

不是 AIMessage。

这条链可以这样理解：

prompt：生成 messages
model：生成 AIMessage
parser：取出文本内容

所以第七天最重要的基础代码就是：

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


model = ChatOpenAI(
    model="qwen-plus",
    api_key="你的 BAILIAN_API_KEY",
    base_url="你的 BAILIAN_BASE_URL",
    temperature=0.3,
)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个耐心、专业的 AI 课程助教。"),
    ("human", "请用学生能听懂的话解释：{question}"),
])

chain = prompt | model | StrOutputParser()

answer = chain.invoke({
    "question": "什么是 LangChain？"
})

print(answer)

先把这段跑通。

只要这段跑通，LangChain 的基础就有了。

11. LCEL：用管道组织流程

LangChain 里这种：

prompt | model | parser

可以理解成一种管道。

它的特点是：

左边的输出，成为右边的输入。

举个更明确的例子：

chain = prompt | model | StrOutputParser()

执行时：

1. prompt 收到 {"question": "..."}
2. prompt 输出 messages
3. model 收到 messages
4. model 输出 AIMessage
5. StrOutputParser 收到 AIMessage
6. StrOutputParser 输出字符串

所以 LangChain 的 Chain 不是神秘东西。

它就是：

一组按顺序执行的步骤。

如果学生理解了这一点，后面的 RunnableLambda、RunnableParallel、RAG Chain 都会容易很多。

12. Runnable 是什么

在 LangChain 里，很多东西都可以被当成 Runnable。

可以先把 Runnable 理解成 LangChain 里的一种统一接口：

接收输入
处理输入
返回输出

也就是说，只要一个组件能按照 LangChain 规定的方式运行，它就可以叫 Runnable。

如果用 Java 来类比，它有点像一个统一的处理接口。

比如 Java 里可以写一个 Function：

Function<String, String> step = question -> {
    return "回答：" + question;
};

String result = step.apply("什么是 RAG？");

这里的 Function 有一个固定规则：

输入一个值
返回一个值

LangChain 里的 Runnable 也有类似思想：

result = runnable.invoke(input_data)

它也有一个固定规则：

输入 input_data
返回 result

区别是，LangChain 的 Runnable 不只支持一次普通调用，还统一支持：

比如：

1. ChatPromptTemplate 是 Runnable。
2. ChatOpenAI 模型对象是 Runnable。
3. StrOutputParser 是 Runnable。
4. 自己的函数也可以包装成 Runnable。

这就是为什么它们可以用 | 串起来。

因为每一步都遵守同一套调用规则。

输入
  ↓
Runnable
  ↓
输出

再看这条链：

chain = prompt | model | StrOutputParser()

它能串起来，是因为每一步都是 Runnable：

prompt 是 Runnable：输入 dict，输出 messages
model 是 Runnable：输入 messages，输出 AIMessage
parser 是 Runnable：输入 AIMessage，输出字符串

所以 | 的意思就是：

把上一步的输出，交给下一步的 Runnable。

LangChain 的流程编排能力，就是建立在 Runnable 这个统一接口上的。

13. RunnableLambda：接入自己的 Python 函数

这是第七天非常重要的一章。

因为我们的项目里已经有很多自己的函数和服务：

1. 查询历史消息。
2. 判断是否需要知识库。
3. 调用 LlamaIndex 检索资料。
4. 筛选可用工具。
5. 查询课表。
6. 转换业务数据格式。

如果引入 LangChain 就要全部重写，那成本太高。

好在不需要。

LangChain 可以用 RunnableLambda 把普通 Python 函数包装进链路。

例如：

from langchain_core.runnables import RunnableLambda


def prepare_context(input_data: dict) -> dict:
    question = input_data["question"]

    return {
        "question": question,
        "context": "这里是根据问题查到的知识库资料",
    }


context_loader = RunnableLambda(prepare_context)

然后就可以把它放进链路：

chain = context_loader | prompt | model | StrOutputParser()

调用：

answer = chain.invoke({
    "question": "什么是 RAG？"
})

执行过程是：

1. 输入 {"question": "什么是 RAG？"}
2. prepare_context 查询或准备上下文
3. 输出 {"question": "...", "context": "..."}
4. prompt 使用 question 和 context 生成 messages
5. model 调用大模型
6. parser 得到字符串

这对项目特别有用。

因为我们可以把已有的服务接进去：

def prepare_context(input_data: dict) -> dict:
    question = input_data["question"]
    history = input_data.get("history", [])

    knowledge_hits = rag_service.retrieve_by_question(question)
    context = format_knowledge_hits(knowledge_hits)

    return {
        "question": question,
        "history": history,
        "context": context,
    }

这里要记住：

LangChain 不是要求我们推翻旧代码，而是让旧代码可以进入统一链路。

14. RunnableParallel：同时准备多个变量

有时候 Prompt 需要多个变量。

比如：

question：学生本次问题
context：知识库资料
history：历史消息
student_info：学生信息

这些变量可能来自不同地方：

1. question 来自用户输入。
2. context 来自知识库检索。
3. history 来自 Java 查出来的数据库聊天记录。
4. student_info 来自登录态或业务接口。

如果不用 LangChain，一步一步手写，大概会变成：

question = input_data["question"]
context = search_knowledge(question)
history = load_history(input_data["conversation_id"])
student_info = load_student_info(input_data["student_id"])

这段代码做的事情其实是：

从同一份 input_data 里，同时准备多个字段。

RunnableParallel 就是用来表达这件事的。

它会把同一个输入分发给多个 Runnable：

                  ┌─ 生成 question
input_data ───────┼─ 生成 context
                  ├─ 生成 history
                  └─ 生成 student_info

最后合并成一个 dict

所以 RunnableParallel 的重点不是“写法更炫”，而是：

把 Prompt 需要的多个变量，一次性准备好。

LangChain 里可以这样表达：

from langchain_core.runnables import RunnableParallel, RunnableLambda


def get_question(input_data: dict) -> str:
    return input_data["question"]


def search_knowledge(input_data: dict) -> str:
    question = input_data["question"]
    return f"根据问题 `{question}` 查到的知识库资料"


def load_history_text(input_data: dict) -> str:
    conversation_id = input_data["conversation_id"]
    return f"会话 {conversation_id} 的历史消息"


def load_student_info_text(input_data: dict) -> str:
    student_id = input_data["student_id"]
    return f"学生 {student_id} 的基础信息"


question_chain = RunnableLambda(get_question)
context_chain = RunnableLambda(search_knowledge)
history_chain = RunnableLambda(load_history_text)
student_info_chain = RunnableLambda(load_student_info_text)

parallel = RunnableParallel({
    "question": question_chain,
    "context": context_chain,
    "history": history_chain,
    "student_info": student_info_chain,
})

这段代码的意思是：

我要准备 4 个变量。

question      交给 question_chain 来生成
context       交给 context_chain 来生成
history       交给 history_chain 来生成
student_info  交给 student_info_chain 来生成

如果用普通 Python 代码来理解，它大概等价于：

result = {
    "question": question_chain.invoke(input_data),
    "context": context_chain.invoke(input_data),
    "history": history_chain.invoke(input_data),
    "student_info": student_info_chain.invoke(input_data),
}

也就是说：

字典的 key，是最后输出结果里的字段名。
字典的 value，是负责生成这个字段的 Runnable。

14.1 RunnableParallel

RunnableParallel 的意思是：

把同一个输入交给多个 Runnable。
每个 Runnable 负责生成一个字段。
最后把这些字段合并成一个字典。

例如：

input_data = {
    "question": "什么是 RAG？",
    "conversation_id": 1001,
    "student_id": 2001,
}

parallel.invoke(input_data)

可能得到：

{
    "question": "什么是 RAG？",
    "context": "根据问题 `什么是 RAG？` 查到的知识库资料",
    "history": "会话 1001 的历史消息",
    "student_info": "学生 2001 的基础信息",
}

注意这里的关系：

输入是一个 dict：input_data
输出也是一个 dict：Prompt 需要的变量

也就是说，RunnableParallel 常用来做这件事：

把业务请求数据，整理成 Prompt 可以直接使用的数据。

这样就刚好可以传给 Prompt：

rag_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是 AI 课程助教，请优先参考知识库资料回答。"),
    ("system", "知识库资料：{context}"),
    ("system", "历史消息：{history}"),
    ("system", "学生信息：{student_info}"),
    ("human", "{question}"),
])

rag_chain = parallel | rag_prompt | model | StrOutputParser()

调用：

answer = rag_chain.invoke({
    "question": "什么是 RAG？",
    "conversation_id": 1001,
    "student_id": 2001,
})

14.2 RunnablePassthrough

RunnablePassthrough 的意思是：

原样把输入传下去。

它适合用在输入本身就可以直接保留的场景。

比如输入就是一个问题字符串：

from langchain_core.runnables import RunnableParallel, RunnablePassthrough, RunnableLambda


def search_knowledge(question: str) -> str:
    return "这里是知识库资料"


parallel = RunnableParallel({
    "question": RunnablePassthrough(),
    "context": RunnableLambda(search_knowledge),
})

调用：

parallel.invoke("什么是 RAG？")

结果是：

{
    "question": "什么是 RAG？",
    "context": "这里是知识库资料",
}

这里的 RunnablePassthrough() 就是把原始问题保留下来，放到 question 字段里。

这个写法很适合解释：

RAG 不是模型自己查资料。
RAG 是程序先查资料，再把资料放进 Prompt。

15. 在 LangChain 里接入现有 LlamaIndex 检索

这里不需要重新学习 LlamaIndex。

前面已经知道：

LlamaIndex + Milvus 负责知识库检索。

现在重点看：

怎么把检索结果接到 LangChain 链路里。

可以先写一个普通函数：

def search_knowledge(question: str) -> str:
    # 这里调用现有 LlamaIndex / Milvus 检索逻辑。
    # 真实项目里可以调用 ChatRagService 或 KnowledgeService。
    hits = rag_service.search(question)

    chunks = []
    for index, hit in enumerate(hits, start=1):
        chunks.append(
            f"[资料{index}] 文档：{hit.document_name}\n{hit.content}"
        )

    return "\n\n".join(chunks)

这段代码里要注意：

1. 函数输入是用户问题。
2. 函数内部调用知识库检索。
3. 函数输出是适合放进 Prompt 的文本。

然后包装成 Runnable：

from langchain_core.runnables import RunnableLambda, RunnableParallel, RunnablePassthrough


knowledge_runnable = RunnableLambda(search_knowledge)

定义 RAG Prompt：

from langchain_core.prompts import ChatPromptTemplate


rag_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个耐心、专业的 AI 课程助教。"),
    (
        "system",
        "下面是从知识库检索到的参考资料。"
        "回答时优先依据资料；资料没有提到的内容，不要编造。\n\n{context}"
    ),
    ("human", "{question}"),
])

组合：

rag_chain = (
    RunnableParallel({
        "question": RunnablePassthrough(),
        "context": knowledge_runnable,
    })
    | rag_prompt
    | model
    | StrOutputParser()
)

这里为什么要写：

"question": RunnablePassthrough()

因为这条链调用时传入的是一个字符串：

"课程里的 RAG 是怎么工作的？"

RunnableParallel 里面有两路：

question 这一路：把原始问题保留下来，给 Prompt 的 {question} 使用
context  这一路：拿原始问题去查知识库，生成 Prompt 的 {context}

如果不写 RunnablePassthrough()，后面的 Prompt 就拿不到原始问题：

("human", "{question}")

所以这里的“原样输出”不是没用，而是为了把用户问题继续传给 Prompt。

调用：

answer = rag_chain.invoke("课程里的 RAG 是怎么工作的？")

这条链的执行过程是：

输入问题
  ↓
RunnableParallel 同时准备 question 和 context
  ↓
context 来自 LlamaIndex 检索结果
  ↓
Prompt 把资料和问题组装成 messages
  ↓
Model 生成回答
  ↓
Parser 输出字符串

这里要讲清楚一个边界：

LangChain 没有替代 LlamaIndex。
LlamaIndex 继续负责知识库检索。
LangChain 负责把检索结果放进整体流程。

这样前后课程就不会冲突。

16. 历史消息怎么接进链路

项目里对话记忆已经存在数据库里。

所以第七天不讲“怎么把聊天记录存数据库”。

只讲：

从数据库查出来的历史消息，怎么进入 LangChain Prompt。

假设 Java 已经传来了历史消息：

history = [
    {"role": "user", "content": "什么是 RAG？"},
    {"role": "assistant", "content": "RAG 是检索增强生成。"},
    {"role": "user", "content": "它和普通问答有什么区别？"},
]

可以转换成 LangChain Message：

from langchain_core.messages import HumanMessage, AIMessage


def convert_history(history: list[dict]):
    messages = []
    for item in history:
        if item["role"] == "user":
            messages.append(HumanMessage(content=item["content"]))
        elif item["role"] == "assistant":
            messages.append(AIMessage(content=item["content"]))
    return messages

然后在 Prompt 里使用 MessagesPlaceholder：

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder


chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个耐心、专业的 AI 课程助教。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{question}"),
])

调用：

history_messages = convert_history(history)

chain = chat_prompt | model | StrOutputParser()

answer = chain.invoke({
    "history": history_messages,
    "question": "再用简单一点的话解释一下。",
})

这里这里要记住：

历史消息不是模型自己记住的。
历史消息是程序从数据库查出来，再放回 messages。

MessagesPlaceholder 的作用是：

在 Prompt 模板里预留一段位置，用来放多条历史消息。

如果不用它，历史消息就只能手动插入。

用了它，Prompt 结构会更清楚：

system 规则
历史消息
本次问题

17. 把历史消息和知识库资料一起放进链路

真实 AI 问答通常不只是 RAG，也不只是多轮对话。

它经常需要：

1. 看懂学生本次问题。
2. 参考历史上下文。
3. 需要时参考知识库资料。

可以定义一个 Prompt：

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder


assistant_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个耐心、专业的 AI 课程助教。"),
    (
        "system",
        "下面是知识库资料。回答时优先参考资料；"
        "如果资料没有明确说明，请告诉学生资料里没有明确说明。\n\n{context}"
    ),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{question}"),
])

准备输入：

input_data = {
    "question": "它和普通问答有什么区别？",
    "history": convert_history(history),
}

准备上下文函数：

def prepare_assistant_context(input_data: dict) -> dict:
    question = input_data["question"]
    history = input_data.get("history", [])

    context = search_knowledge(question)

    return {
        "question": question,
        "history": history,
        "context": context,
    }

组合：

assistant_chain = (
    RunnableLambda(prepare_assistant_context)
    | assistant_prompt
    | model
    | StrOutputParser()
)

调用：

answer = assistant_chain.invoke(input_data)

这条链对应真实项目里的核心问答流程。

这里要注意：

实际项目不一定马上重构成这条链。
我们先学会这种组织方式，后面再决定哪些地方值得引入。

18. Tool API：把业务能力声明给 AI

第五天已经讲过工具调用。

这里重点看 LangChain 里的 Tool API。

注意：

Tool 不是 Agent。

Tool 是一个可调用的业务能力。

Agent 是会选择工具的执行流程。

可以这样定义一个工具：

from langchain_core.tools import tool


@tool
def get_student_day_schedule(date: str) -> str:
    """查询当前学生某一天的课程安排。"""
    return f"{date} 有 Python AI 项目课。"

这个函数有三个关键信息：

1. 函数名：get_student_day_schedule
2. 参数：date
3. 文档注释：查询当前学生某一天的课程安排。

文档注释非常重要。

因为模型会根据工具说明判断：

这个工具能不能解决用户的问题？
应该给这个工具传什么参数？

如果工具说明写得模糊，模型就容易选错工具。

例如不好的说明：

@tool
def get_student_day_schedule(date: str) -> str:
    """查信息。"""
    ...

这个说明太模糊。

更好的说明：

@tool
def get_student_day_schedule(date: str) -> str:
    """查询当前登录学生在指定日期的课程安排，date 格式为 yyyy-MM-dd。"""
    ...

这样模型更容易知道：

1. 什么时候用这个工具。
2. 参数应该是什么格式。
3. 这个工具只能查当前登录学生，不应该让用户指定 studentId。

19. 工具设计里的安全边界

在学生端 AI 问答里，工具不能随便设计。

比如查课表工具，不应该这样写：

@tool
def get_student_day_schedule(student_id: int, date: str) -> str:
    """查询某个学生某一天的课程安排。"""
    ...

这个设计有风险。

因为模型可能从用户问题里提取出别人的学生 ID。

甚至用户可能故意说：

帮我查 studentId=10086 的课表。

正确思路是：

学生身份来自 Java 已鉴权上下文。
模型只负责提取日期。

所以工具应该设计成：

@tool
def get_student_day_schedule(date: str) -> str:
    """查询当前登录学生在指定日期的课程安排，date 格式为 yyyy-MM-dd。"""
    ...

真正执行工具时，后端从上下文里拿当前学生 ID：

def execute_schedule_tool(current_student_id: int, date: str) -> str:
    # current_student_id 来自 Java 鉴权后的学生上下文
    # date 来自模型提取的工具参数
    return schedule_service.get_day_schedule(current_student_id, date)

这里要记住：

模型可以帮我们提取参数，但不能替代权限校验。

20. create_agent：先认识 Agent 长什么样

LangChain 也提供了创建 Agent 的 API。

这一节先知道它大概长什么样，不需要马上掌握完整执行流程。

Agent 的细节会放到第八天继续学习。

示例：

from langchain.agents import create_agent


agent = create_agent(
    model=model,
    tools=[get_student_day_schedule],
    system_prompt="你是学生端 AI 问答助手。回答时要简洁、清楚，适合学生理解。",
)

调用：

result = agent.invoke({
    "messages": [
        {
            "role": "user",
            "content": "我今天上什么课？"
        }
    ]
})

这里先记住一个结论：

Tool 是可以被调用的能力。
Agent 是会根据问题选择工具的流程。

第八天再继续解决这些问题：

1. Agent 如何判断是否需要工具。
2. Agent 如何选择工具。
3. Agent 如何填工具参数。
4. 工具执行结果如何回到模型。
5. 为什么工具调用通常会有多次模型请求。

21. 一个完整的小 Demo

下面写一个小 demo，把前面学到的内容串起来。

这个 demo 不直接改项目主流程。

它只演示：

LangChain 怎么组织 Prompt、模型、知识库上下文和输出解析。

21.1 创建模型

from langchain_openai import ChatOpenAI


model = ChatOpenAI(
    model="qwen-plus",
    api_key="你的 BAILIAN_API_KEY",
    base_url="你的 BAILIAN_BASE_URL",
    temperature=0.3,
    max_tokens=1000,
)

21.2 模拟知识库检索

为了先看清楚 LangChain 的链路，这里先用 mock 数据模拟知识库检索。

def search_knowledge(question: str) -> str:
    if "RAG" in question or "知识库" in question:
        return (
            "[资料1] RAG 是 Retrieval-Augmented Generation 的缩写，"
            "中文通常叫检索增强生成。它会先从外部资料中检索相关内容，"
            "再把资料和用户问题一起交给大模型生成回答。"
        )

    return "知识库中没有找到明确相关资料。"

等 demo 跑通以后，再替换成真实项目里的检索逻辑：

真实项目里，这个函数内部可以调用 LlamaIndex / Milvus。

21.3 定义 Prompt

from langchain_core.prompts import ChatPromptTemplate


prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个耐心、专业的 AI 课程助教。"),
    (
        "system",
        "下面是知识库资料。回答时优先依据资料；"
        "如果资料没有说明，不要编造。\n\n{context}"
    ),
    ("human", "{question}"),
])

21.4 组织 Chain

from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda, RunnableParallel, RunnablePassthrough


knowledge_runnable = RunnableLambda(search_knowledge)

chain = (
    RunnableParallel({
        "question": RunnablePassthrough(),
        "context": knowledge_runnable,
    })
    | prompt
    | model
    | StrOutputParser()
)

21.5 调用 Chain

answer = chain.invoke("什么是 RAG？")

print(answer)

这个 demo 的执行过程是：

输入："什么是 RAG？"
  ↓
RunnableParallel 准备两个变量
  ├─ question：原样保留用户问题
  └─ context：调用 search_knowledge 查询资料
  ↓
ChatPromptTemplate 生成 messages
  ↓
ChatOpenAI 调用模型
  ↓
StrOutputParser 输出字符串

这个 demo 就是第七天的核心。

能看懂这个执行过程，就已经理解 LangChain 的主要用法了。