Skip to content

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            │ 字节开源    │ 组件化/流式编排        │  │
│  └────────────────┴──────────────┴───────────────────────┘  │
└──────────────────────────────────────────────────────────────┘

框架对比 ​

框架语言定位核心特点适合场景
LangChainPython全栈框架组件丰富、Chain 编排快速原型、通用开发
LangGraphPython图编排状态图、循环、多 Agent复杂工作流、多 Agent
LlamaIndexPythonRAG 框架检索优化、数据连接知识库、搜索引擎
CrewAIPython多 Agent角色扮演、任务委派模拟团队协作
AutoGenPython多 Agent对话式协作、代码执行代码生成、研究助手
LangChain4jJavaJava 生态Spring Boot 友好企业 Java 应用
Spring AIJavaSpring 生态与 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 ChainPrompt + LLM + Parser简单问答
RAG ChainRetriever + Prompt + LLM检索增强生成
Conversation ChainMemory + 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-openaiOpenAI 集成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 稳定丢失早期信息长对话
SummaryLLM 摘要信息保留+省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 领域最深                    │                    │
│  └──────────────────────────────────────┘                    │
└──────────────────────────────────────────────────────────────┘

详细对比 ​

维度LangChainLlamaIndex
核心定位通用编排框架数据/RAG 框架
核心抽象Runnable/ChainIndex/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)
语言PythonJava
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 复杂                   │              │
│  └────────────────────────────────────────────┘              │
└──────────────────────────────────────────────────────────────┘

核心区别 ​

维度LangChainLangGraph
执行模型线性管道(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
特性旧版 ChainLCEL
流式❌✅ 原生
异步部分✅ 原生
批处理❌✅ 原生
类型提示弱✅ 强
可组合性有限✅ 任意组合
调试难✅ 回调追踪

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 Researcho1 驱动,数十分钟深度调研
Gemini Deep ResearchGoogle 搜索 + Gemini 综合
Perplexity Pro实时搜索 + 引用
自建方案LangGraph + 搜索 API + LLM

追问延伸:

  • Deep Research 和 RAG 有什么区别?(RAG 从已有知识库检索;Deep Research 实时搜索网络)
  • 如何控制 Deep Research 的成本?(限制迭代次数 + 小模型预筛 + 缓存搜索结果)