Appearance
LangChain 框架面试专题
LangChain 是 AI Agent 开发的主流框架。主线:Chain 理解 → 架构原理 → Agent 构建 → 记忆机制 → 框架对比 → LangGraph → 工程实践。
Q1: 你了解过哪些 AI Agent 开发框架? 「🟢 校招/初级」
考察点:AI Agent 开发框架的全景了解。
参考答案:
┌──────────────────────────────────────────────────────────────┐
│ AI Agent 开发框架全景 │
│ │
│ Python 生态 │
│ ┌────────────────┬──────────────┬───────────────────────┐ │
│ │ LangChain │ 最全面 │ Chain/Agent/Memory/RAG │ │
│ │ LangGraph │ 图编排 │ 状态图/多Agent/循环 │ │
│ │ LlamaIndex │ RAG 专精 │ 检索/索引/查询引擎 │ │
│ │ CrewAI │ 多Agent │ 角色协作/任务分配 │ │
│ │ AutoGen │ 多Agent │ 微软出品/对话式协作 │ │
│ │ Semantic Kernel │ 微软企业级 │ 技能/规划/记忆 │ │
│ └────────────────┴──────────────┴───────────────────────┘ │
│ │
│ Java 生态 │
│ ┌────────────────┬──────────────┬───────────────────────┐ │
│ │ LangChain4j │ Java版LC │ Chain/Agent/Memory │ │
│ │ Spring AI │ Spring生态 │ Spring Boot集成 │ │
│ └────────────────┴──────────────┴───────────────────────┘ │
│ │
│ Go 生态 │
│ ┌────────────────┬──────────────┬───────────────────────┐ │
│ │ LangChainGo │ Go版LC │ 轻量/高性能 │ │
│ │ Eino │ 字节开源 │ 组件化/流式编排 │ │
│ └────────────────┴──────────────┴───────────────────────┘ │
└──────────────────────────────────────────────────────────────┘框架对比
| 框架 | 语言 | 定位 | 核心特点 | 适合场景 |
|---|---|---|---|---|
| LangChain | Python | 全栈框架 | 组件丰富、Chain 编排 | 快速原型、通用开发 |
| LangGraph | Python | 图编排 | 状态图、循环、多 Agent | 复杂工作流、多 Agent |
| LlamaIndex | Python | RAG 框架 | 检索优化、数据连接 | 知识库、搜索引擎 |
| CrewAI | Python | 多 Agent | 角色扮演、任务委派 | 模拟团队协作 |
| AutoGen | Python | 多 Agent | 对话式协作、代码执行 | 代码生成、研究助手 |
| LangChain4j | Java | Java 生态 | Spring Boot 友好 | 企业 Java 应用 |
| Spring AI | Java | Spring 生态 | 与 Spring 无缝集成 | Spring 项目 |
python
# LangChain 最简 Agent
from langchain.agents import create_react_agent, AgentExecutor
from langchain.tools import tool
from langchain_openai import ChatOpenAI
@tool
def search(query: str) -> str:
"""搜索网络信息"""
return f"搜索结果: {query}"
llm = ChatOpenAI(model="gpt-4")
agent = create_react_agent(llm, [search])
executor = AgentExecutor(agent=agent, tools=[search])
result = executor.invoke({"input": "今天北京天气"})追问延伸:
- 为什么 Java 生态的框架较少?(AI 生态在 Python,Java 框架通常是移植版,滞后于原版)
- 选框架时最看重什么?(生态活跃度、文档质量、社区支持、性能)
Q2: 如何理解 LangChain 中的 Chain? 「🟢 校招/初级」
考察点:LangChain Chain 概念的理解。
参考答案:
Chain 是 LangChain 中将多个组件串联起来执行特定任务的基本单元,类似于函数组合(Function Composition)。
┌──────────────────────────────────────────────────────────────┐
│ Chain 的本质 │
│ │
│ Chain = 组件 A → 组件 B → 组件 C → 输出 │
│ │
│ 类比: │
│ Unix 管道: cat file | grep "pattern" | sort | uniq │
│ LangChain: Prompt | LLM | OutputParser | Result │
│ │
│ 每个 Chain 接收输入 → 经过一系列组件 → 产出输出 │
└──────────────────────────────────────────────────────────────┘Chain 的演进
LangChain v0.1(旧版 LCEL 之前):
from langchain.chains import LLMChain
chain = LLMChain(llm=llm, prompt=prompt)
result = chain.run("问题")
LangChain v0.2+(LCEL - LangChain Expression Language):
chain = prompt | llm | output_parser
result = chain.invoke({"question": "问题"})
# LCEL 的优势:
# 1. 更 Pythonic(管道运算符)
# 2. 原生支持流式输出
# 3. 异步支持(ainvoke)
# 4. 批处理(batch)
# 5. 可组合(Chain 可以嵌套)常见 Chain 类型
| Chain 类型 | 组成 | 用途 |
|---|---|---|
| LLM Chain | Prompt + LLM + Parser | 简单问答 |
| RAG Chain | Retriever + Prompt + LLM | 检索增强生成 |
| Conversation Chain | Memory + Prompt + LLM | 多轮对话 |
| Sequential Chain | 多个 Chain 串联 | 多步处理 |
| Router Chain | 分类器 + 多个 Chain | 动态路由 |
python
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
# LCEL 基础 Chain
prompt = ChatPromptTemplate.from_template("回答问题:{question}")
llm = ChatOpenAI(model="gpt-4")
parser = StrOutputParser()
# 管道组合
chain = prompt | llm | parser
# 调用
result = chain.invoke({"question": "什么是RAG?"})
# prompt.format → llm.invoke → parser.parse
# 流式调用
for chunk in chain.stream({"question": "什么是RAG?"}):
print(chunk, end="", flush=True)
# 批处理
results = chain.batch([
{"question": "什么是RAG?"},
{"question": "什么是Agent?"},
{"question": "什么是MCP?"}
])
# 异步调用
result = await chain.ainvoke({"question": "什么是RAG?"})RAG Chain 示例
python
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
# 构建 RAG Chain
retriever = vector_store.as_retriever(search_kwargs={"k": 5})
prompt = ChatPromptTemplate.from_template("""
基于以下上下文回答问题:
上下文:{context}
问题:{question}
""")
# LCEL 组合 RAG
rag_chain = (
{"context": retriever, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
result = rag_chain.invoke("什么是向量数据库?")追问延伸:
- LCEL 相比旧版 Chain 有什么优势?(流式、异步、批处理、可组合性、更好的类型提示)
- Chain 能嵌套吗?(可以,一个 Chain 的输出可以作为另一个 Chain 的输入)
Q3: LangChain 的底层架构与实现原理是什么? 「🟡 中级」
考察点:LangChain 架构设计的理解。
参考答案:
┌──────────────────────────────────────────────────────────────┐
│ LangChain 架构层级 │
│ │
│ ┌────────────────────────────────────────────────────┐ │
│ │ 应用层 (Applications) │ │
│ │ LangGraph / LangSmith / LangServe │ │
│ └────────────────────────┬───────────────────────────┘ │
│ │ │
│ ┌────────────────────────┴───────────────────────────┐ │
│ │ 编排层 (Orchestration) │ │
│ │ LCEL (Runnable) / Agents / Workflows │ │
│ └────────────────────────┬───────────────────────────┘ │
│ │ │
│ ┌────────────────────────┴───────────────────────────┐ │
│ │ 组件层 (Components) │ │
│ │ Prompts / LLMs / Tools / Memory / Retrievers │ │
│ │ Output Parsers / Document Loaders / Splitters │ │
│ └────────────────────────┬───────────────────────────┘ │
│ │ │
│ ┌────────────────────────┴───────────────────────────┐ │
│ │ 集成层 (Integrations) │ │
│ │ langchain-openai / langchain-anthropic / etc │ │
│ │ langchain-community (第三方集成) │ │
│ └────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘核心抽象:Runnable 协议
python
# Runnable 是 LangChain 的核心抽象
# 所有组件都实现 Runnable 接口
from langchain_core.runnables import Runnable
class Runnable:
"""所有 LangChain 组件的基类"""
def invoke(self, input, config=None):
"""同步调用"""
raise NotImplementedError
async def ainvoke(self, input, config=None):
"""异步调用"""
raise NotImplementedError
def stream(self, input, config=None):
"""流式输出"""
raise NotImplementedError
def batch(self, inputs, config=None):
"""批量处理"""
raise NotImplementedError
# 运算符重载
def __or__(self, other):
"""管道运算符 | 实现链式组合"""
return RunnableSequence(self, other)核心模块
| 模块 | 职责 | 核心接口 |
|---|---|---|
| langchain-core | 基础抽象 | Runnable, PromptValue, BaseMessage |
| langchain | 高级编排 | Agent, Chain, Memory |
| langchain-community | 第三方集成 | 各种 LLM/向量库/工具 |
| langchain-openai | OpenAI 集成 | ChatOpenAI, OpenAIEmbeddings |
| langchain-experimental | 实验功能 | CSV Agent, PAL |
| langgraph | 图编排 | StateGraph, 节点/边 |
LCEL 实现原理
python
# LCEL 的管道运算符实现
class RunnableSequence(Runnable):
"""管道组合的实现"""
def __init__(self, first, last):
self.first = first # 前一个 Runnable
self.last = last # 后一个 Runnable
def invoke(self, input, config=None):
# 依次执行每个组件
result = self.first.invoke(input, config)
return self.last.invoke(result, config)
def stream(self, input, config=None):
# 流式:前一个的输出作为后一个的输入
for chunk in self.first.stream(input, config):
yield from self.last.stream(chunk, config)
async def ainvoke(self, input, config=None):
result = await self.first.ainvoke(input, config)
return await self.last.ainvoke(result, config)
# chain = prompt | llm | parser
# 等价于:
# chain = RunnableSequence(
# RunnableSequence(prompt, llm),
# parser
# )事件回调系统
python
from langchain_core.callbacks import BaseCallbackHandler
class MyCallbackHandler(BaseCallbackHandler):
def on_llm_start(self, serialized, prompts, **kwargs):
print(f"LLM 开始: {prompts[0][:50]}")
def on_llm_end(self, response, **kwargs):
print(f"LLM 结束: 用时 {response.llm_output}")
def on_tool_start(self, tool, input_str, **kwargs):
print(f"工具调用: {tool} - {input_str}")
def on_tool_end(self, output, **kwargs):
print(f"工具结果: {output}")
# 全局回调
chain.invoke("问题", config={"callbacks": [MyCallbackHandler()]})追问延伸:
- LangChain 为什么要拆分这么多包?(解耦,核心包稳定,集成包快速迭代)
- Runnable 协议的设计灵感来自什么?(Java 的 Stream API、RxJava 的 Observable)
Q4: 使用 LangChain 构建 Agent 的核心步骤是什么? 「🟢 校招/初级」
考察点:LangChain Agent 构建的实践。
参考答案:
┌──────────────────────────────────────────────────────────────┐
│ LangChain 构建 Agent 的核心步骤 │
│ │
│ Step 1: 选择 LLM │
│ ┌──────────────────────────────────┐ │
│ │ llm = ChatOpenAI(model="gpt-4") │ │
│ └──────────────────────────────────┘ │
│ │ │
│ Step 2: 定义工具 │
│ ┌──────────────────────────────────┐ │
│ │ @tool │ │
│ │ def search(query: str) -> str: │ │
│ │ """搜索网络""" │ │
│ │ return results │ │
│ └──────────────────────────────────┘ │
│ │ │
│ Step 3: 创建 Agent │
│ ┌──────────────────────────────────┐ │
│ │ agent = create_react_agent( │ │
│ │ llm=llm, │ │
│ │ tools=[search, calculator] │ │
│ │ ) │ │
│ └──────────────────────────────────┘ │
│ │ │
│ Step 4: 包装执行器 │
│ ┌──────────────────────────────────┐ │
│ │ executor = AgentExecutor( │ │
│ │ agent=agent, │ │
│ │ tools=[search, calculator], │ │
│ │ memory=memory, │ │
│ │ verbose=True │ │
│ │ ) │ │
│ └──────────────────────────────────┘ │
│ │ │
│ Step 5: 执行任务 │
│ ┌──────────────────────────────────┐ │
│ │ result = executor.invoke({ │ │
│ │ "input": "用户问题" │ │
│ │ }) │ │
│ └──────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘python
from langchain.agents import create_react_agent, AgentExecutor
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from langchain_core.prompts import ChatPromptTemplate
# Step 1: 选择 LLM
llm = ChatOpenAI(model="gpt-4", temperature=0)
# Step 2: 定义工具
@tool
def search_web(query: str) -> str:
"""搜索网络信息,输入搜索关键词"""
# 实际调用搜索 API
return f"搜索 '{query}' 的结果:..."
@tool
def calculate(expression: str) -> str:
"""数学计算,输入数学表达式"""
try:
result = eval(expression) # 实际项目用安全的计算器
return str(result)
except:
return "计算错误"
tools = [search_web, calculate]
# Step 3: 创建 Agent(ReAct 范式)
prompt = ChatPromptTemplate.from_template("""
你是一个有用的助手,可以使用以下工具:
{tools}
工具名称: {tool_names}
问题: {input}
思考: {agent_scratchpad}
""")
agent = create_react_agent(llm, tools, prompt)
# Step 4: 包装执行器
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
executor = AgentExecutor(
agent=agent,
tools=tools,
memory=memory,
verbose=True,
max_iterations=5, # 防止无限循环
handle_parsing_errors=True
)
# Step 5: 执行
result = executor.invoke({
"input": "3的5次方是多少?再搜索一下这个数字有什么含义"
})
print(result["output"])Agent 类型对比
| Agent 类型 | 说明 | 适用 |
|---|---|---|
| ReAct | 推理+行动交替 | 通用任务 |
| Structured Tool Chat | 工具输出结构化 | 复杂工具 |
| OpenAI Functions | 原生函数调用 | GPT 模型 |
| Self-Ask | 自我提问分解 | 知识检索 |
| Plan-and-Execute | 先规划再执行 | 复杂任务 |
追问延伸:
- 如何控制 Agent 的迭代次数?(max_iterations 参数 + 超时控制)
- Agent 执行失败怎么处理?(handle_parsing_errors + 回退策略 + 人工兜底)
Q5: 在 LangChain 中,如何为 Agent 注册工具? 「🟢 校招/初级」
考察点:LangChain 工具注册方式。
参考答案:
方式一:@tool 装饰器(推荐)
python
from langchain.tools import tool
@tool
def search(query: str) -> str:
"""搜索网络信息
Args:
query: 搜索关键词
"""
return search_api(query)
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""发送邮件
Args:
to: 收件人邮箱
subject: 邮件主题
body: 邮件正文
"""
return smtp.send(to, subject, body)
tools = [search, send_email]方式二:继承 BaseTool 类
python
from langchain.tools import BaseTool
from typing import Optional, Type
from pydantic import BaseModel, Field
class SearchInput(BaseModel):
"""搜索工具的输入模型"""
query: str = Field(description="搜索关键词")
max_results: int = Field(default=10, description="最大结果数")
class SearchTool(BaseTool):
"""自定义搜索工具"""
name: str = "search"
description: str = "搜索网络信息"
args_schema: Type[BaseModel] = SearchInput
def _run(self, query: str, max_results: int = 10) -> str:
"""同步执行"""
return search_api(query, max_results)
async def _arun(self, query: str, max_results: int = 10) -> str:
"""异步执行"""
return await async_search_api(query, max_results)
tools = [SearchTool()]方式三:StructuredTool.from_function
python
from langchain.tools import StructuredTool
def search_func(query: str, max_results: int = 10) -> str:
"""搜索函数"""
return search_api(query, max_results)
search_tool = StructuredTool.from_function(
func=search_func,
name="search",
description="搜索网络信息",
# 可选:自定义参数 Schema
args_schema=SearchInput,
)
tools = [search_tool]方式四:使用内置工具
python
from langchain_community.tools import (
DuckDuckGoSearchRun, # 搜索
PythonREPL, # Python 代码执行
ShellTool, # Shell 命令
WolframAlphaQueryRun, # 数学/科学查询
)
tools = [
DuckDuckGoSearchRun(),
PythonREPL(),
]工具注册最佳实践
| 实践 | 说明 |
|---|---|
| 描述要清晰 | description 是 LLM 选择工具的依据 |
| 参数有说明 | 每个参数都要有 description |
| 返回值统一 | 返回字符串,便于 LLM 理解 |
| 错误处理 | 工具内部 try-catch,返回错误信息 |
| 异步支持 | 实现 _arun 方法 |
| 类型注解 | 使用 Pydantic/类型注解 |
python
# 好的工具定义
@tool
def query_database(
sql: str,
database: str = "default"
) -> str:
"""执行 SQL 查询并返回结果
当用户需要查询数据库时使用此工具。
只支持 SELECT 语句,不支持 DDL/DML。
Args:
sql: SQL 查询语句(仅 SELECT)
database: 数据库名称,默认为 default
Returns:
JSON 格式的查询结果
"""
if not sql.strip().upper().startswith("SELECT"):
return "错误:只支持 SELECT 查询"
try:
result = db.execute(sql, database)
return json.dumps(result, ensure_ascii=False)
except Exception as e:
return f"查询失败:{str(e)}"追问延伸:
- 工具太多时怎么管理?(工具检索:Embedding 工具描述,动态选择相关工具)
- 如何限制 Agent 使用危险工具?(权限控制 + 人工审批 + 沙箱执行)
Q6: LangChain 如何实现短期记忆和长期记忆? 「🟡 中级」
考察点:LangChain 记忆机制的实现。
参考答案:
┌──────────────────────────────────────────────────────────────┐
│ LangChain 记忆体系 │
│ │
│ 短期记忆 (Conversation Memory) │
│ ┌────────────────────────────────────────────┐ │
│ │ 存储方式: 内存中(进程级) │ │
│ │ 内容: 当前对话历史 │ │
│ │ 生命周期: 会话级 │ │
│ │ 实现: ConversationBufferMemory 等 │ │
│ └────────────────────────────────────────────┘ │
│ │
│ 长期记忆 (Persistent Memory) │
│ ┌────────────────────────────────────────────┐ │
│ │ 存储方式: 数据库/向量库/文件 │ │
│ │ 内容: 跨会话的知识和事实 │ │
│ │ 生命周期: 永久 │ │
│ │ 实现: 向量存储 + 检索 │ │
│ └────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘短期记忆实现
python
from langchain.memory import (
ConversationBufferMemory, # 完整对话历史
ConversationBufferWindowMemory, # 滑动窗口
ConversationSummaryMemory, # 摘要记忆
ConversationTokenBufferMemory, # Token 限制记忆
)
# 1. 完整历史(简单但占 Token)
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True # 返回消息列表
)
# 2. 滑动窗口(只保留最近 K 轮)
memory = ConversationBufferWindowMemory(
k=5, # 保留最近 5 轮对话
memory_key="chat_history"
)
# 3. 摘要记忆(LLM 生成摘要,省 Token)
memory = ConversationSummaryMemory(
llm=llm,
memory_key="chat_history"
# LLM 自动将旧对话总结为摘要
)
# 4. Token 限制(按 Token 数截断)
memory = ConversationTokenBufferMemory(
llm=llm,
max_token_limit=2000, # 最多 2000 Token
memory_key="chat_history"
)短期记忆对比
| 记忆类型 | 原理 | 优点 | 缺点 | 适用 |
|---|---|---|---|---|
| Buffer | 全量保存 | 信息完整 | Token 消耗大 | 短对话 |
| Window | 只存最近 K 轮 | Token 稳定 | 丢失早期信息 | 长对话 |
| Summary | LLM 摘要 | 信息保留+省Token | 摘要有损 | 超长对话 |
| Token Buffer | 按 Token 截断 | 精确控制 | 截断不完整 | 上下文受限 |
长期记忆实现
python
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
class LongTermMemory:
"""基于向量检索的长期记忆"""
def __init__(self):
self.vector_store = Chroma(
embedding_function=OpenAIEmbeddings(),
collection_name="long_term_memory"
)
def add(self, content: str, metadata: dict = None):
"""添加记忆"""
self.vector_store.add_texts(
texts=[content],
metadatas=[metadata or {}]
)
def search(self, query: str, k: int = 5) -> list:
"""检索相关记忆"""
return self.vector_store.similarity_search(query, k=k)
def update(self, old_id: str, new_content: str):
"""更新记忆"""
self.vector_store.delete([old_id])
self.vector_store.add_texts([new_content])
# 使用:短期 + 长期记忆组合
class HybridMemory:
"""混合记忆:短期对话 + 长期知识"""
def __init__(self, llm):
self.short_term = ConversationBufferWindowMemory(k=10)
self.long_term = LongTermMemory()
def load_memory(self, query: str) -> str:
"""加载记忆上下文"""
# 短期:最近对话
recent = self.short_term.buffer_as_str
# 长期:检索相关记忆
long_term = self.long_term.search(query, k=3)
long_term_str = "\n".join([d.page_content for d in long_term])
return f"最近对话:\n{recent}\n\n相关记忆:\n{long_term_str}"
def save(self, human_msg: str, ai_msg: str):
"""保存对话"""
# 短期保存
self.short_term.save_context(
{"input": human_msg},
{"output": ai_msg}
)
# 重要信息提取后存入长期记忆
important = self._extract_important(human_msg, ai_msg)
if important:
self.long_term.add(important)追问延伸:
- 如何判断什么信息该存入长期记忆?(用 LLM 判断信息重要性,或基于规则提取实体/事实)
- 记忆系统如何做压缩?(摘要 + 实体提取 + 时间衰减 + 检索筛选)
Q7: LangChain 和 LlamaIndex 有什么区别? 「🟡 中级」
考察点:两大主流框架的对比。
参考答案:
┌──────────────────────────────────────────────────────────────┐
│ LangChain vs LlamaIndex │
│ │
│ LangChain │
│ ┌──────────────────────────────────────┐ │
│ │ 定位: 通用 LLM 应用开发框架 │ │
│ │ 核心概念: Chain(链式组合) │ │
│ │ 强项: Agent / 工具调用 / 复杂编排 │ │
│ │ 数据处理: 基础文档加载和切分 │ │
│ │ 检索: 通用检索器接口 │ │
│ │ 生态: 最大社区,最多集成 │ │
│ └──────────────────────────────────────┘ │
│ │
│ LlamaIndex │
│ ┌──────────────────────────────────────┐ │
│ │ 定位: 数据驱动的 LLM 应用框架 │ │
│ │ 核心概念: Index(索引结构) │ │
│ │ 强项: RAG / 文档处理 / 检索优化 │ │
│ │ 数据处理: 丰富索引/分块/查询引擎 │ │
│ │ 检索: 深度检索优化(多路/重排/子查询) │ │
│ │ 生态: RAG 领域最深 │ │
│ └──────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘详细对比
| 维度 | LangChain | LlamaIndex |
|---|---|---|
| 核心定位 | 通用编排框架 | 数据/RAG 框架 |
| 核心抽象 | Runnable/Chain | Index/Query Engine |
| Agent 能力 | ✅ 强(多种范式) | 一般(有但不是重点) |
| RAG 能力 | 基础 | ✅ 最强 |
| 数据连接 | 基础 Loader | ✅ 丰富(100+ 数据源) |
| 索引结构 | 简单 | ✅ 多种(树/图/关键词) |
| 检索优化 | 基础 | ✅ 深度(子查询/重排/融合) |
| 文档处理 | 简单切分 | ✅ 多策略(语义/层级/句子) |
| 流式支持 | ✅ LCEL 原生 | 有 |
| 社区规模 | ✅ 最大 | 大 |
| 适合场景 | Agent/复杂工作流 | RAG/知识库 |
选型建议
场景1: 构建 AI Agent(工具调用 + 多步推理)
→ LangChain / LangGraph
场景2: 企业知识库问答(RAG)
→ LlamaIndex(检索优化最深)
场景3: 两者都要
→ LlamaIndex 做检索 + LangChain 做 Agent 编排
→ 或用 LangChain 的 RAG(够用就行)
场景4: 简单聊天 + 记忆
→ LangChain(ConversationChain 足够)
场景5: 复杂文档处理(PDF/表格/层级)
→ LlamaIndex(文档解析最强)python
# LangChain RAG(简单但够用)
from langchain_community.vectorstores import FAISS
from langchain_openai import ChatOpenAI
retriever = vector_store.as_retriever()
chain = (
{"context": retriever, "question": RunnablePassthrough()}
| prompt | llm | parser
)
# LlamaIndex RAG(检索优化更深)
from llama_index.core import VectorStoreIndex, Settings
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.postprocessor import SentenceEmbeddingOptimizer
# 精细控制
Settings.node_parser = SentenceSplitter(
chunk_size=512, chunk_overlap=50
)
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine(
similarity_top_k=10, # 先多检索
node_postprocessors=[
SentenceEmbeddingOptimizer(percentile_cutoff=0.5) # 再优化
]
)追问延伸:
- 两个框架能混用吗?(可以,LlamaIndex 的检索器可包装为 LangChain 的 Retriever)
- 还有其他 RAG 框架吗?(Haystack、txtai、DSPy 等)
Q8: LangChain4j 主要解决了哪些问题? 「🟡 中级」
考察点:LangChain4j 在 Java 生态中的定位。
参考答案:
LangChain4j 是 LangChain 的 Java 移植版,为 Java 开发者提供 LLM 应用开发能力。
┌──────────────────────────────────────────────────────────────┐
│ LangChain4j 解决的核心问题 │
│ │
│ 问题1: Java 生态缺少 LLM 开发框架 │
│ ┌──────────────────────────────────────────┐ │
│ │ 解决: 提供 LangChain 的 Java 等价能力 │ │
│ │ - Chain / Agent / Memory / RAG │ │
│ │ - 工具调用 / 文档处理 │ │
│ └──────────────────────────────────────────┘ │
│ │
│ 问题2: Java 企业级开发体验 │
│ ┌──────────────────────────────────────────┐ │
│ │ 解决: 与 Java 生态深度集成 │ │
│ │ - Spring Boot 自动配置 │ │
│ │ - 类型安全(Java 泛型) │ │
│ │ - 熟悉的 API 风格 │ │
│ └──────────────────────────────────────────┘ │
│ │
│ 问题3: 多模型/多供应商管理 │
│ ┌──────────────────────────────────────────┐ │
│ │ 解决: 统一的 LLM 接口 │ │
│ │ - OpenAI / Anthropic / Ollama │ │
│ │ - 阿里通义 / 百度千帆 / Amazon Bedrock │ │
│ └──────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘LangChain4j 核心 API
java
import dev.langchain4j.service.*;
import dev.langchain4j.model.openai.OpenAiChatModel;
import dev.langchain4j.memory.chat.MessageWindowChatMemory;
import dev.langchain4j.data.segment.*;
import dev.langchain4j.store.embedding.*;
// 1. 简单对话
ChatLanguageModel model = OpenAiChatModel.builder()
.apiKey("sk-xxx")
.modelName("gpt-4")
.temperature(0.7)
.build();
String answer = model.generate("什么是 RAG?");
// 2. 带记忆的 AI Service(声明式)
interface Assistant {
String chat(String userMessage);
}
Assistant assistant = AiServices.builder(Assistant.class)
.chatLanguageModel(model)
.chatMemory(MessageWindowChatMemory.withMaxMessages(20))
.build();
assistant.chat("我叫张三");
assistant.chat("我叫什么名字?"); // → "你叫张三"
// 3. 带工具的 Agent
@Tool("搜索网络信息")
String search(String query) {
return searchApi(query);
}
interface Agent {
String execute(@MemoryId String userId, @UserMessage String task);
}
Agent agent = AiServices.builder(Agent.class)
.chatLanguageModel(model)
.tools(this) // 自动扫描 @Tool 注解方法
.chatMemoryProvider(userId ->
MessageWindowChatMemory.withMaxMessages(50))
.build();
agent.execute("user1", "帮我搜索 RAG 最新进展");LangChain4j vs LangChain (Python)
| 维度 | LangChain (Python) | LangChain4j (Java) |
|---|---|---|
| 语言 | Python | Java |
| API 风格 | 函数式 + 管道 | 接口 + 注解(声明式) |
| 集成 | 最多 | 核心集成 |
| Spring | 不支持 | ✅ 深度集成 |
| 类型安全 | 动态类型 | 编译时检查 |
| 性能 | 中 | 略好(JVM) |
| 部署 | 需 Python 运行时 | JAR 部署 |
java
// Spring Boot 集成示例
@SpringBootApplication
public class App {
@Bean
ChatLanguageModel model() {
return OpenAiChatModel.builder()
.apiKey(System.getenv("OPENAI_API_KEY"))
.modelName("gpt-4")
.build();
}
@Bean
Assistant assistant(ChatLanguageModel model) {
return AiServices.builder(Assistant.class)
.chatLanguageModel(model)
.build();
}
}
@RestController
class Controller {
@Autowired Assistant assistant;
@PostMapping("/chat")
String chat(@RequestBody String message) {
return assistant.chat(message);
}
}追问延伸:
- LangChain4j 和 Spring AI 怎么选?(Spring AI 更贴近 Spring 生态,LangChain4j 功能更全更独立)
- Java 做 LLM 应用的优势在哪?(企业级稳定性、类型安全、现有系统集成)
Q9: LangChain 和 LangGraph 的核心区别是什么? 「🔴 高级」
考察点:理解从 LangChain 到 LangGraph 的演进。
参考答案:
┌──────────────────────────────────────────────────────────────┐
│ LangChain vs LangGraph │
│ │
│ LangChain │
│ ┌────────────────────────────────────────────┐ │
│ │ 核心: Chain(线性管道) │ │
│ │ 执行: 顺序执行,A→B→C │ │
│ │ 循环: Agent 内部有限循环 │ │
│ │ 状态: 隐式(Memory) │ │
│ │ 适合: 简单 Agent / RAG / 问答 │ │
│ │ 局限: 难以表达复杂控制流 │ │
│ └────────────────────────────────────────────┘ │
│ │
│ LangGraph │
│ ┌────────────────────────────────────────────┐ │
│ │ 核心: StateGraph(状态图) │ │
│ │ 执行: 图遍历,支持分支/循环/并行 │ │
│ │ 循环: 一等公民,任意复杂循环 │ │
│ │ 状态: 显式(State 对象) │ │
│ │ 适合: 多 Agent / 复杂工作流 / 人机协作 │ │
│ │ 局限: 比直接写 Chain 复杂 │ │
│ └────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘核心区别
| 维度 | LangChain | LangGraph |
|---|---|---|
| 执行模型 | 线性管道(DAG) | 状态图(支持循环) |
| 控制流 | 顺序 + 条件分支 | 顺序 + 分支 + 循环 + 并行 |
| 状态管理 | 隐式(Memory) | 显式(TypedDict State) |
| Agent 架构 | 单 Agent + 工具 | 多 Agent + 协作 |
| 循环支持 | 有限(Agent 内部) | 原生支持(图边) |
| 人机协作 | 不支持 | ✅ interrupt() |
| 持久化 | Memory | ✅ Checkpoint |
| 可视化 | 无 | ✅ 图可视化 |
| 适合 | 简单场景 | 复杂 Agent 系统 |
LangGraph 代码示例
python
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
# 1. 定义状态
class AgentState(TypedDict):
messages: Annotated[list, operator.add] # 消息累加
next: str # 下一个节点
# 2. 定义节点(每个节点是一个函数)
def research_agent(state: AgentState) -> dict:
"""研究 Agent"""
response = research_llm.invoke(state["messages"])
return {"messages": [response]}
def draft_agent(state: AgentState) -> dict:
"""写作 Agent"""
response = draft_llm.invoke(state["messages"])
return {"messages": [response]}
def critic_agent(state: AgentAgent) -> dict:
"""审查 Agent"""
response = critic_llm.invoke(state["messages"])
# 决定是否需要修改
if "需要修改" in response.content:
return {"next": "draft"}
return {"next": END}
def router(state: AgentState) -> str:
"""路由:决定下一步"""
return state.get("next", "research")
# 3. 构建图
graph = StateGraph(AgentState)
# 添加节点
graph.add_node("research", research_agent)
graph.add_node("draft", draft_agent)
graph.add_node("critic", critic_agent)
# 添加边(定义执行顺序)
graph.set_entry_point("research")
graph.add_edge("research", "draft")
graph.add_edge("draft", "critic")
# 条件边:critic 可以回到 draft 或结束
graph.add_conditional_edges(
"critic",
router,
{
"draft": "draft", # 需要修改 → 重写
END: END, # 满意 → 结束
}
)
# 4. 编译并执行
app = graph.compile()
result = app.invoke({"messages": [HumanMessage("写一篇关于 RAG 的技术文章")]})LangChain 无法做而 LangGraph 能做的
1. 多 Agent 循环协作
Agent A → Agent B → Agent C → 判断 → 回到 Agent B(循环)
2. 人机协作(Human-in-the-loop)
Agent 执行到某步 → 暂停等待人工审批 → 继续
3. 状态持久化
中途保存状态 → 可恢复执行
4. 并行分支
同时执行多个 Agent → 汇总结果
5. 复杂条件路由
根据状态灵活路由到不同处理路径追问延伸:
- LangGraph 和传统工作流引擎(如 Airflow)有什么区别?(LangGraph 专注 LLM 编排,状态包含消息/对话)
- LangGraph 能替代 LangChain 吗?(简单场景 LangChain 更轻量,复杂场景 LangGraph 更合适)
Q10: LangGraph 相比于 LangChain 有哪些核心优势? 「🔴 高级」
考察点:LangGraph 的优势和应用价值。
参考答案:
优势一:循环和复杂控制流
python
# LangChain:Agent 内部循环受限
# 难以实现"A→B→判断→A 或 C→B→结束"这种复杂循环
# LangGraph:循环是一等公民
graph.add_conditional_edges(
"validator",
lambda state: "revise" if state["needs_revision"] else "done",
{"revise": "writer", "done": END}
)
# writer → validator → writer → validator → done
# 无限循环直到满意优势二:显式状态管理
python
# LangChain:状态隐式在 Memory 中,难以精细控制
# LangGraph:状态是显式的 TypedDict
class WorkflowState(TypedDict):
user_query: str
retrieved_docs: list
draft_answer: str
critique: str
revision_count: int # 显式跟踪修订次数
final_answer: str
# 每个节点读取和更新状态
def writer(state: WorkflowState) -> dict:
return {
"draft_answer": llm.generate(state["user_query"], state["retrieved_docs"]),
"revision_count": state["revision_count"] + 1
}优势三:多 Agent 协作
python
from langgraph.graph import StateGraph, END
class TeamState(TypedDict):
task: str
research_result: str
draft: str
review: str
approved: bool
# 多个专业 Agent
def researcher(state): ...
def writer(state): ...
def reviewer(state): ...
def editor(state): ...
graph = StateGraph(TeamState)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
graph.add_node("reviewer", reviewer)
graph.add_node("editor", editor)
graph.set_entry_point("researcher")
graph.add_edge("researcher", "writer")
graph.add_edge("writer", "reviewer")
graph.add_conditional_edges(
"reviewer",
lambda s: "writer" if not s["approved"] else "editor",
{"writer": "writer", "editor": "editor"}
)
graph.add_edge("editor", END)
# researcher → writer → reviewer → (不通过) → writer → reviewer → (通过) → editor → END优势四:人机协作
python
from langgraph.checkpoint.memory import MemorySaver
# 编译时添加 checkpointer(支持暂停/恢复)
app = graph.compile(
checkpointer=MemorySaver(),
interrupt_before=["editor"] # 在 editor 前暂停
)
# 执行到 editor 前自动暂停
config = {"configurable": {"thread_id": "thread-1"}}
result = app.invoke({"task": "写文章"}, config)
# 人工审核后继续
result = app.invoke(None, config) # None 表示从暂停处继续优势五:状态持久化和恢复
python
from langgraph.checkpoint.sqlite import SqliteSaver
# 使用 SQLite 持久化状态
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
app = graph.compile(checkpointer=checkpointer)
# 即使进程重启,也能从断点恢复
# 通过 thread_id 恢复之前的执行状态
result = app.invoke(
{"task": "继续之前的任务"},
{"configurable": {"thread_id": "thread-1"}}
)优势总结
| 优势 | 说明 | 价值 |
|---|---|---|
| 循环控制 | 原生支持复杂循环 | 迭代优化、多轮审查 |
| 显式状态 | TypedDict 管理状态 | 可调试、可追踪 |
| 多 Agent | 图节点 = Agent | 角色分工、协作 |
| 人机协作 | interrupt 机制 | 人工审批、安全控制 |
| 持久化 | Checkpoint | 断点恢复、长任务 |
| 可视化 | 图结构可视化 | 理解和调试复杂流程 |
| 并行 | 支持并行分支 | 提高效率 |
追问延伸:
- LangGraph 适合替代传统工作流引擎吗?(专注 AI 场景,通用工作流还是用 Airflow/Temporal)
- 多 Agent 通信开销怎么控制?(共享状态 vs 消息传递,按需选择)
Q11: LangChain 大版本升级有哪些核心变化? 「🟡 中级」
考察点:LangChain 版本演进的理解。
参考答案:
┌──────────────────────────────────────────────────────────────┐
│ LangChain 版本演进 │
│ │
│ v0.1 (2024.1) │
│ ┌──────────────────────────────────┐ │
│ │ - 引入 LCEL (LangChain Expression Language)│ │
│ │ - 包拆分: langchain-core / community / openai│ │
│ │ - Runnable 协议统一接口 │ │
│ │ - 弃用旧版 Chain API │ │
│ └──────────────────────────────────┘ │
│ │
│ v0.2 (2024.5) │
│ ┌──────────────────────────────────┐ │
│ │ - langchain-core 独立发布 │ │
│ │ - 社区集成独立包 │ │
│ │ - 更好的异步支持 │ │
│ │ - Structured Output 原生支持 │ │
│ │ - 工具定义改进(@tool 增强) │ │
│ └──────────────────────────────────┘ │
│ │
│ v0.3 (2024.9+) │
│ ┌──────────────────────────────────┐ │
│ │ - 移除大量旧 API(legacy 清理) │ │
│ │ - Agent 标准化(create_react_agent)│ │
│ │ - 更好的 Pydantic v2 支持 │ │
│ │ - 性能优化 │ │
│ │ - LangGraph 深度集成 │ │
│ └──────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘核心变化详解
1. LCEL(最重要的变化)
python
# 旧版 (v0.0.x)
from langchain.chains import LLMChain
chain = LLMChain(llm=llm, prompt=prompt)
result = chain.run("问题")
# 新版 LCEL (v0.1+)
chain = prompt | llm | parser
result = chain.invoke({"question": "问题"})
# 支持 stream/batch/ainvoke| 特性 | 旧版 Chain | LCEL |
|---|---|---|
| 流式 | ❌ | ✅ 原生 |
| 异步 | 部分 | ✅ 原生 |
| 批处理 | ❌ | ✅ 原生 |
| 类型提示 | 弱 | ✅ 强 |
| 可组合性 | 有限 | ✅ 任意组合 |
| 调试 | 难 | ✅ 回调追踪 |
2. 包拆分
python
# 旧版:一个大包
pip install langchain
from langchain.llms import OpenAI # ❌ 已弃用
from langchain.chat_models import ChatOpenAI # ❌ 已弃用
# 新版:模块化
pip install langchain langchain-openai langchain-community
from langchain_openai import ChatOpenAI # ✅
from langchain_community.vectorstores import FAISS # ✅
from langchain_core.prompts import ChatPromptTemplate # ✅3. Agent API 变化
python
# 旧版:多种 Agent 类型,API 不统一
from langchain.agents import AgentType, initialize_agent
agent = initialize_agent(
tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION
)
# 新版:统一为 create_react_agent
from langchain.agents import create_react_agent, AgentExecutor
agent = create_react_agent(llm, tools)
executor = AgentExecutor(agent=agent, tools=tools)4. 工具定义变化
python
# 旧版
from langchain.tools import Tool
tool = Tool(name="search", func=search_func, description="搜索")
# 新版
from langchain.tools import tool
@tool
def search(query: str) -> str:
"""搜索"""
return search_func(query)迁移建议
| 场景 | 建议 |
|---|---|
| 新项目 | 直接用 v0.3 + LCEL |
| 旧项目 | 逐步迁移,旧 API 仍可用但有警告 |
| Agent | 用 create_react_agent |
| RAG | 用 LCEL 组合 |
| 复杂工作流 | 直接用 LangGraph |
追问延伸:
- 旧版 Chain 还能用吗?(可以但会有 deprecation warning,未来会移除)
- LCEL 和 LangGraph 的关系?(LCEL 适合简单管道,LangGraph 适合复杂图编排,可以混用)
Q12: Deep Research 的实现逻辑和适用场景是什么? 「🔴 高级」
考察点:Deep Research 这一新兴 Agent 模式的理解。
参考答案:
Deep Research 是一种多轮、迭代式的 Agent 工作模式,通过反复搜索→分析→推理→再搜索的循环,对复杂问题进行深度调研并生成长篇研究报告。
┌──────────────────────────────────────────────────────────────┐
│ Deep Research 工作流程 │
│ │
│ 用户问题 → 规划阶段 → 执行阶段 → 综合阶段 → 报告输出 │
│ │
│ ┌─────────────────────────────────────────────┐ │
│ │ 规划阶段 │ │
│ │ 1. 分析用户问题,确定调研目标 │ │
│ │ 2. 分解为子问题/搜索关键词 │ │
│ │ 3. 制定搜索计划 │ │
│ └────────────────────┬────────────────────────┘ │
│ ▼ │
│ ┌─────────────────────────────────────────────┐ │
│ │ 执行阶段(循环) │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐│ │
│ │ │ 搜索 │──→│ 分析 │──→│ 推理 ││ │
│ │ │ Web/API │ │ 提取关键 │ │ 补充搜索 ││ │
│ │ │ │←──│ 信息 │←──│ 新问题 ││ │
│ │ └──────────┘ └──────────┘ └──────────┘│ │
│ │ ←────── 迭代循环 ──────→ │ │
│ └────────────────────┬────────────────────────┘ │
│ ▼ │
│ ┌─────────────────────────────────────────────┐ │
│ │ 综合阶段 │ │
│ │ 1. 汇总所有搜索结果 │ │
│ │ 2. 去重/交叉验证 │ │
│ │ 3. 组织报告结构 │ │
│ │ 4. 生成带引用的详细报告 │ │
│ └─────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘实现逻辑
python
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
class ResearchState(TypedDict):
question: str # 原始问题
search_plan: list # 搜索计划
search_results: Annotated[list, operator.add] # 累加搜索结果
analysis: str # 分析结果
new_queries: list # 新发现的子问题
report: str # 最终报告
iteration: int # 迭代次数
# 节点1: 规划
def planner(state: ResearchState) -> dict:
prompt = f"""
分析以下问题,制定搜索计划:
问题:{state['question']}
输出:3-5 个搜索关键词,按优先级排列。
"""
plan = llm.invoke(prompt)
return {"search_plan": parse_keywords(plan)}
# 节点2: 搜索
def searcher(state: ResearchState) -> dict:
results = []
for query in state["search_plan"]:
result = web_search(query)
results.append({"query": query, "content": result})
return {"search_results": results}
# 节点3: 分析
def analyzer(state: ResearchState) -> dict:
prompt = f"""
基于搜索结果分析问题:
问题:{state['question']}
搜索结果:{state['search_results']}
1. 提取关键信息
2. 标记信息缺口
3. 提出需要补充搜索的问题
"""
analysis = llm.invoke(prompt)
new_queries = extract_new_queries(analysis)
return {
"analysis": analysis,
"new_queries": new_queries,
"iteration": state["iteration"] + 1
}
# 节点4: 判断是否继续
def should_continue(state: ResearchState) -> str:
if state["iteration"] >= 5: # 最多 5 轮
return "synthesize"
if not state["new_queries"]: # 没有新问题了
return "synthesize"
return "search_again"
# 节点5: 综合报告
def synthesizer(state: ResearchState) -> dict:
prompt = f"""
基于所有搜索和分析结果,生成详细研究报告:
原始问题:{state['question']}
所有结果:{state['search_results']}
分析过程:{state['analysis']}
要求:
1. 结构化报告(摘要/正文/结论)
2. 标注引用来源
3. 客观中立
4. 2000-5000字
"""
report = llm.invoke(prompt)
return {"report": report}
# 构建图
graph = StateGraph(ResearchState)
graph.add_node("planner", planner)
graph.add_node("searcher", searcher)
graph.add_node("analyzer", analyzer)
graph.add_node("synthesizer", synthesizer)
graph.set_entry_point("planner")
graph.add_edge("planner", "searcher")
graph.add_edge("searcher", "analyzer")
graph.add_conditional_edges(
"analyzer",
should_continue,
{
"search_again": "searcher", # 继续搜索
"synthesize": "synthesizer", # 生成报告
}
)
graph.add_edge("synthesizer", END)
deep_research = graph.compile()适用场景
| 场景 | 说明 | 效果 |
|---|---|---|
| 技术调研 | 调研某技术的原理/对比/选型 | 自动搜索多源信息+对比分析 |
| 市场研究 | 竞品分析/行业趋势 | 多维度信息汇总 |
| 学术综述 | 某领域的文献综述 | 多轮检索+交叉验证 |
| 新闻深挖 | 某事件的深度报道 | 多角度搜索+时间线梳理 |
| 产品分析 | 用户需求/市场分析 | 综合多数据源 |
Deep Research 的关键挑战
| 挑战 | 说明 | 方案 |
|---|---|---|
| 搜索质量 | 搜索结果可能噪声多 | 多路搜索 + Reranking |
| 信息验证 | 搜索到的信息可能不准确 | 交叉验证多个来源 |
| 上下文限制 | 大量搜索结果超出上下文窗口 | 分批处理 + 摘要压缩 |
| 成本 | 多轮搜索 + LLM 调用 | 控制迭代次数 + 小模型预筛 |
| 延迟 | 多轮迭代耗时长 | 异步并行 + 缓存 |
| 报告质量 | 综合多源信息生成报告 | 结构化模板 + 引用标注 |
代表产品
| 产品 | 特点 |
|---|---|
| OpenAI Deep Research | o1 驱动,数十分钟深度调研 |
| Gemini Deep Research | Google 搜索 + Gemini 综合 |
| Perplexity Pro | 实时搜索 + 引用 |
| 自建方案 | LangGraph + 搜索 API + LLM |
追问延伸:
- Deep Research 和 RAG 有什么区别?(RAG 从已有知识库检索;Deep Research 实时搜索网络)
- 如何控制 Deep Research 的成本?(限制迭代次数 + 小模型预筛 + 缓存搜索结果)