1. 这不是框架选型指南，而是大模型工程落地的“操作系统”图谱

你刚接触大模型开发时，是不是也经历过这种混乱：看到别人用 LangChain 写了个问答系统，自己照着跑通了，但加个搜索功能就卡在 Retriever 配置上；想试试更“现代”的 LangGraph，结果光是搞懂 StateGraph 和 add_messages 的类型注解就花了两天；听说 LlamaIndex 在文档处理上更顺手，可一查发现它和 LangChain 的 Document 对象不兼容，又得重写数据加载逻辑……这不是你学得慢，而是当前整个生态里，没有一张清晰、真实、带血丝的“框架地图”——所有教程都默认你已经站在山顶往下看，却没人告诉你，从山脚出发时，哪条路坑最多、哪段坡最陡、哪个岔口一旦走错就得原路返回三小时。

我做 LLM 应用落地项目六年，亲手交付过 23 个生产级 RAG 系统、7 个复杂 Agent 工作流、4 套企业级知识中枢平台。这期间踩过的坑，90% 都出在框架层的选择与混用上。LangChain 不是“万能胶”，LlamaIndex 也不是“文档神器”，LangGraph 更不是“下一代 LangChain”。它们本质是不同抽象层级的“操作系统内核”：LangChain 是面向任务编排的“进程调度器”，LlamaIndex 是面向非结构化数据的“文件系统驱动”，LangGraph 是面向状态演化的“内核态线程管理器”，而 LangFuse 则是贯穿全栈的“性能探针与诊断仪”。今天这篇，不讲 API 怎么调，不列参数怎么填，只做一件事：把这七个核心组件——LangChain、LlamaIndex、LangGraph、Agently、LangFuse、LiteLLM、LangFlow——放在真实工程场景的显微镜下，拆开看它们的内存布局、中断响应机制、上下文切换成本，以及最关键的： 你在什么时刻、为什么必须用它，又在什么临界点上必须立刻弃用它 。关键词不是“入门”或“教程”，而是“决策边界”、“替换成本”、“可观测性缺口”。如果你正卡在“该用哪个框架”的十字路口，或者已经上线却总在深夜被 CallbackHandler 的 trace 断点搞崩溃，那这篇就是为你写的实战地形图。

2. 框架本质解构：不是功能对比，而是运行时角色定位

2.1 LangChain：大模型应用的“Shell 脚本解释器”

很多人误以为 LangChain 是“链式调用工具库”，这是致命误解。它的核心设计哲学，是把 LLM 当作一个不可信、不可控、高延迟的外部命令（类似 curl 或 ssh ），而 LangChain 就是那个帮你写 bash 脚本、管理环境变量、处理 stderr/stdout、做重试和超时的 shell 解释器。所以你看它的核心抽象： Chain 是脚本文件， LLMChain 是执行 curl 的那一行， Agent 是带 if-else 和 for 循环的完整脚本， Tool 是你提前写好的 .sh 函数库。它不关心模型内部怎么推理，只关心“调用是否成功”“返回是否 JSON”“错误码是不是 429”。

这就决定了它的适用边界： 当你需要快速组合已有能力（搜索、计算、API 调用）并容忍一定黑盒性时，LangChain 是最快路径；但当你需要精确控制 token 流、干预中间状态、或做细粒度错误恢复时，它会成为瓶颈 。比如 ConversationalRetrievalChain ，表面看是“对话+检索”，实际运行时，它把用户问题、历史消息、检索结果一股脑塞给 LLM，你根本无法在检索后、生成前插入校验逻辑——因为整个流程被封装在一个 run() 方法里，就像你不能在 curl -X POST 执行中途修改请求体。

提示：LangChain 的 Runnable 接口是重大进化，但它仍是“单次调用封装”，不是“状态机”。 RunnableParallel 看似并发，实则只是 asyncio.gather 的语法糖，各分支间无状态共享。别被 with_config() 的灵活性迷惑——配置变更需重建整个 Runnable 实例，生产环境热更新几乎不可行。

2.2 LlamaIndex：非结构化数据的“索引编译器”

LlamaIndex 的名字极具误导性。“Index” 不是指数据库索引，而是“编译索引”（Indexing Compiler）。它把 PDF、Word、网页等原始文档，当作源代码，通过 Document 抽象为 AST（抽象语法树），再经 NodeParser 切片为“语法节点”，最后用 VectorStoreIndex 编译成可查询的向量二进制。这个过程高度类比 C++ 编译： Document 是 .cpp 文件， NodeParser 是预处理器（ #include 展开、宏替换）， EmbeddingModel 是词法分析器， VectorStore 是链接后的 .so 库。所以它的强项从来不是“快”，而是“可控”——你能精确指定 SentenceSplitter 的 chunk_size、overlap，能自定义 MetadataMode 控制哪些元信息参与 embedding，甚至能写 TransformComponent 做全文正则清洗。

但代价是： 它天然排斥“动态数据流”。LlamaIndex 的 QueryEngine 设计假设数据是静态的、批量编译完成的。当你需要实时接入 Kafka 流、处理用户上传的 Excel 并即时检索时，它的 refresh() 接口会触发全量 re-index，延迟秒级起跳 。这也是为什么生产环境常见“LangChain + LlamaIndex”组合：LangChain 做外层工作流调度（接收请求、路由、调用工具），LlamaIndex 做内层数据编译与查询（只对已入库文档生效）。二者不是替代关系，而是“操作系统”与“文件系统”的协作关系。

2.3 LangGraph：状态驱动的“内核态线程管理器”

LangGraph 的颠覆性，在于它把“Agent”从“函数调用链”升维为“状态机进程”。LangChain 的 AgentExecutor 是单线程阻塞式执行：思考→选工具→调用→解析→思考…，所有状态压在 Python 栈帧里，崩溃即丢失。LangGraph 的 StateGraph 则模拟操作系统内核：每个 node 是一个独立进程， state 是共享内存段， conditional_edge 是信号量， interrupt 是硬件中断。你定义 State 类型，就像声明 struct ； add_node 是 fork() 子进程； invoke() 是启动调度器； stream() 是按时间片轮询。所以 add_messages 的 Annotated[list, add_messages] 本质是原子操作： state["messages"] += new_msg ，而非简单追加——它确保多 node 并发写入时不会丢消息。

这就解释了为什么 LangGraph 必须强制类型注解：它在编译期就生成状态迁移图（类似 Linux ftrace 的 sched_switch 图），runtime 只做状态校验与跳转。因此， LangGraph 的学习曲线陡峭，但收益是确定性的：你可以用 get_state() 随时 dump 进程内存，用 update_state() 注入调试指令，甚至用 interrupt() 暂停正在运行的 classify_email node，手动修改 is_spam 标志再 resume——这在 LangChain 里需要重写整个 AgentExecutor 。它的适用场景非常明确：需要长期记忆（如客服对话）、多步骤协同（如电商下单：查库存→扣减→发券→通知）、或需人工介入的审批流。

2.4 Agently：轻量级 Agent 的“嵌入式 RTOS”

Agently 是生态里的异类。它没有 pip install agently 的官方包，核心是一个 300 行的 agent.py ，设计理念直指 LangChain/LangGraph 的臃肿： 去掉所有抽象层，用纯 Python 字典和函数实现 Agent 的最小可行内核 。它的 Agent 类就是一个 dict ， set_llm() 是注入一个 callable， run() 就是 while True: 执行 step() 。没有 Runnable ，没有 StateGraph ，没有 CallbackHandler ——所有可观测性靠 print() 和 logging 。这看似简陋，实则是精准打击：当你的需求是“给销售团队做个微信小程序里的产品问答机器人”，要求 2 小时上线、零运维、月活 < 500，Agently 的 agent.run("推荐一款适合程序员的机械键盘") 比 LangChain 的 LLMChain 启动更快、内存占用更低、debug 更直观。

注意：Agently 的“轻量”是双刃剑。它不提供 Retriever 、不集成向量库、不支持异步。你得自己写 search_knowledge_base(query) 函数，并保证它返回格式符合 agent.set_context() 。它的价值不在功能，而在“无框架心智负担”——当你连 pip install 都想省掉时，Agently 就是答案。

2.5 LangFuse：全栈可观测性的“eBPF 探针”

LangFuse 不是“日志收集器”，而是 LLM 应用的 eBPF。它不依赖应用主动打日志，而是通过 CallbackHandler 注入到 LangChain/LangGraph 的执行钩子中，像 eBPF 程序一样在 kernel space 捕获每一次 llm.invoke() 的输入输出、token 数、耗时、甚至中间 tool_call 的参数。它的 Trace 是进程树， Span 是线程栈帧， Score 是 perf event。所以 langfuse.score_trace(name="user-feedback", value=1) 不是写数据库，而是向 tracing ring buffer 发送一个事件。这解释了为什么 LangFuse 能做“无侵入监控”：你无需改一行业务代码，只要在 invoke() 的 config={"callbacks": [langfuse_handler]} 里加上 handler，整个调用链就自动被 instrumented。

但这也带来关键限制： LangFuse 的观测粒度，完全取决于你使用的框架是否暴露了足够深的 hook。LangChain 的 Runnable 支持 fine-grained span，但原生 LLMChain 只有一个粗粒度 span；LangGraph 的 stream() 可以捕获每个 node 的执行，但若你用 invoke() 一次性跑完，就只能看到顶层 span 。所以 LangFuse 不是银弹，它是“探针”，而探针能探测到什么，取决于被测程序的 debug symbol 是否完整。

2.6 LiteLLM：模型网关的“OpenResty”

LiteLLM 的本质，是 LLM 世界的 OpenResty。它不训练模型，不处理 prompt，只做三件事：协议转换（将 OpenAI 格式请求转为 Anthropic、Azure、Ollama 等格式）、负载均衡（根据 latency/cost 自动路由到最优 endpoint）、熔断限流（ max_retries=3 + fallbacks=["gpt-3.5-turbo"] ）。它的 completion() 函数，就像 Nginx 的 proxy_pass ，背后可以是 https://api.openai.com ，也可以是 http://localhost:8000/v1 的本地 vLLM。所以当你看到 litellm + langfuse 组合，实际架构是：App → LiteLLM（协议适配/路由）→ LangFuse（流量观测）→ LLM Provider。LiteLLM 的 litellm.success_callback = ["langfuse"] 不是集成，而是让 LiteLLM 主动向 LangFuse 发送 on_success 事件，相当于 OpenResty 的 log_by_lua_block 。

实操心得：LiteLLM 的 model_alias_map 是救命功能。生产环境常需灰度发布新模型，比如 gpt-4o-mini 替换 gpt-4o 。只需在 alias map 里设 "gpt-4o": "gpt-4o-mini" ，所有调用 completion(model="gpt-4o") 的旧代码自动切流，零代码修改。这比改 LangChain 的 ChatOpenAI(model="gpt-4o") 安全十倍。

2.7 LangFlow：可视化编排的“低代码 IDE”

LangFlow 不是“拖拽建站工具”，而是 Jupyter Notebook 的工程化延伸。它的每个组件（ ChatOpenAI 、 ChromaLoader 、 PromptTemplate ）本质是一个预定义的 Python class，画布上的连线就是 input / output 的字典键映射。当你点击“运行”，LangFlow 后端实际是动态生成一个 Python 脚本： from langchain_openai import ChatOpenAI; llm = ChatOpenAI(...); result = llm.invoke(input_data) 。所以它的优势在于“所见即所得调试”：你拖一个 PromptTemplate ，右边实时渲染 template.format(question="xxx") 的结果；连一条线到 ChatOpenAI ，就能看到 messages 数组的结构变化。这极大降低了 RunnableParallel 、 RouterRunnable 等高级组合的理解门槛。

但硬伤同样明显：**LangFlow 的导出代码是“不可维护的”。它生成的 app.py 是扁平化脚本，没有模块划分、没有类型注解、没有错误处理。一个 20 节点的流程图导出后，是 300 行无缩进的 def run(): 函数。所以 LangFlow 的正确定位是“原型验证 IDE”，而非“生产部署框架”。我们团队的标准流程是：用 LangFlow 30 分钟搭出 MVP 流程 → 导出代码 → 重构为 StateGraph + TypedDict → 加入 langfuse 监控 → 部署为 langserve API。LangFlow 省下的不是开发时间，而是跨部门对齐成本——产品经理拖拽完，工程师直接拿代码重构，双方对“流程长什么样”毫无歧义。

3. 混合架构实战：如何用七框架搭建生产级 RAG 系统

3.1 架构全景图：分层解耦的设计哲学

我们以“企业内部知识库问答系统”为例，展示七框架如何各司其职。整个系统分为四层：

• 接入层（LangFlow） ：提供 Web UI，支持非技术人员上传 PDF/Word，配置知识库名称、描述、访问权限。所有操作实时生成 LangFlow 流程图，导出为标准 yaml 配置。
• 编译层（LlamaIndex） ：监听 LangFlow 的上传事件，触发 LlamaIndex 的 VectorStoreIndex.from_documents() 。关键配置： SentenceSplitter(chunk_size=256, chunk_overlap=64) 确保技术文档的代码块不被截断； MetadataMode.ALL 将文件名、页码、章节标题作为 metadata embedding，提升检索相关性。
• 调度层（LangChain + LiteLLM） ：接收用户查询，先用 LangChain 的 MultiQueryRetriever 生成 3 个变体问题（"RAG 是什么" → "请解释 RAG 架构"、"RAG 的优缺点"、"RAG 与微调的区别"），并行检索；结果经 LiteLLM 路由到 gpt-4o-mini （快）或 gpt-4o （准），依据 litellm.max_budget 动态选择。
• 执行层（LangGraph + LangFuse） ：将检索结果、用户问题、历史消息构造成 State ，交由 LangGraph 的 StateGraph 处理。 graph_builder.add_node("answer", answer_node) 中的 answer_node 函数，内部调用 LiteLLM 完成最终生成，并通过 langfuse_handler 记录完整 trace。 conditional_edge 实现“若置信度<0.8，则触发人工审核 node”。

这个架构里，没有“主框架”，只有“角色分工”。LangFlow 不参与运行，只负责配置下发；LlamaIndex 不处理请求，只专注数据编译；LangChain 是胶水，LiteLLM 是网关，LangGraph 是引擎，LangFuse 是仪表盘。任何一层可独立替换：明天要换向量库？只改 LlamaIndex 的 VectorStoreIndex 初始化；后天要接入私有模型？只改 LiteLLM 的 model_list ；大后天要加审批流？只在 LangGraph 里新增 review_node 和 conditional_edge 。

3.2 关键代码片段：从概念到可运行的细节

3.2.1 LlamaIndex 数据编译：解决“PDF 表格识别失真”问题

企业文档常含表格，原生 UnstructuredPDFLoader 会把表格转成混乱文本。我们用 pymupdf4llm 替代：

# requirements.txt
pymupdf4llm==0.0.32
llama-index-core==0.10.42
llama-index-readers-file==0.10.42

# loader.py
import fitz  # PyMuPDF
from llama_index.core import Document
from llama_index.readers.file import PDFReader

class TableAwarePDFReader(PDFReader):
    def load_data(self, file_path: str, extra_info: dict = None) -> list[Document]:
        # 使用 PyMuPDF 精确提取文本和表格
        doc = fitz.open(file_path)
        documents = []
        for page_num in range(len(doc)):
            page = doc[page_num]
            # 提取纯文本（保留换行）
            text = page.get_text("text")
            # 提取表格（返回 pandas DataFrame 列表）
            tables = page.find_tables()
            if tables.tables:
                for i, table in enumerate(tables.tables):
                    # 将表格转为 markdown 格式，避免信息丢失
                    table_md = table.to_markdown()
                    text += f"\n\n| 表格 {i+1} |\n|---|\n{table_md}\n"
            documents.append(
                Document(
                    text=text,
                    metadata={
                        "source": file_path,
                        "page": page_num + 1,
                        "file_name": os.path.basename(file_path),
                    }
                )
            )
        return documents

# 编译入口
from llama_index.core import VectorStoreIndex
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb

client = chromadb.PersistentClient(path="./chroma_db")
collection = client.create_collection("enterprise_knowledge")
vector_store = ChromaVectorStore(chroma_collection=collection)

# 使用自定义 reader
loader = TableAwarePDFReader()
documents = loader.load_data(file_path="./manual.pdf")

index = VectorStoreIndex.from_documents(
    documents,
    vector_store=vector_store,
    transformations=[
        # 关键：使用 SentenceSplitter 精确切分
        SentenceSplitter(chunk_size=256, chunk_overlap=64)
    ]
)

这段代码解决了三个痛点：1）表格内容不丢失；2）metadata 包含页码，便于前端高亮；3）chunk_size=256 确保单个 chunk 不超过 gpt-4o-mini 的 context window 1/10，避免 truncation。

3.2.2 LangGraph 状态机：实现“追问澄清”工作流

用户问“怎么配置 SSO？”，系统需判断是否需澄清（SSO for which system? Okta or Azure AD?）。传统 LangChain 需写复杂 Agent logic，LangGraph 用 conditional_edge 清晰表达：

from typing import TypedDict, List, Optional, Dict, Any
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, SystemMessage

class RAGState(TypedDict):
    messages: Annotated[List[Any], add_messages]
    context: Optional[str]
    needs_clarification: bool
    clarification_question: Optional[str]

def retrieve(state: RAGState) -> Dict[str, Any]:
    """检索知识库，返回 context"""
    user_query = state["messages"][-1].content
    # 使用 LlamaIndex 的 query_engine
    response = index.as_query_engine().query(user_query)
    context = response.response
    # 简单规则判断是否需澄清（实际可用 LLM 分类）
    needs_clarification = "sso" in user_query.lower() and "okta" not in user_query.lower()
    clarification_question = "请问您指的是 Okta 还是 Azure AD 的 SSO 配置？" if needs_clarification else None
    return {
        "context": context,
        "needs_clarification": needs_clarification,
        "clarification_question": clarification_question
    }

def generate_answer(state: RAGState) -> Dict[str, Any]:
    """生成最终回答"""
    prompt = f"""你是一个企业 IT 支持助手。请基于以下知识库内容，准确回答用户问题。
知识库内容：
{state['context']}

用户问题：
{state['messages'][-1].content}

请用中文回答，简洁专业。"""
    # 通过 LiteLLM 调用
    from litellm import completion
    response = completion(
        model="gpt-4o-mini",
        messages=[{"role": "system", "content": "You are an IT support assistant."}, 
                 {"role": "user", "content": prompt}]
    )
    return {"messages": [HumanMessage(content=response.choices[0].message.content)]}

def ask_clarification(state: RAGState) -> Dict[str, Any]:
    """发送澄清问题"""
    return {"messages": [HumanMessage(content=state["clarification_question"])]}

def route_to_clarify(state: RAGState) -> str:
    """路由函数：决定是否澄清"""
    return "clarify" if state["needs_clarification"] else "answer"

# 构建图
graph_builder = StateGraph(RAGState)
graph_builder.add_node("retrieve", retrieve)
graph_builder.add_node("answer", generate_answer)
graph_builder.add_node("clarify", ask_clarification)

graph_builder.add_edge(START, "retrieve")
graph_builder.add_conditional_edges(
    "retrieve",
    route_to_clarify,
    {
        "clarify": "clarify",
        "answer": "answer"
    }
)
graph_builder.add_edge("clarify", "retrieve")  # 澄清后重新检索
graph_builder.add_edge("answer", END)

graph = graph_builder.compile()

这个 StateGraph 的精妙在于： clarify node 发送问题后， graph.stream() 会自然等待用户下一轮输入（ HumanMessage ），然后再次进入 retrieve ，形成闭环。无需 while True ，无需全局变量，状态完全由 State 管理。

3.2.3 LangFuse 全链路监控：捕捉“幻觉”发生的精确位置

LangFuse 的 trace 不仅记录耗时，更能定位幻觉源头。我们在 generate_answer node 中加入 LLM-as-a-Judge：

from langfuse import get_client
from langfuse.decorators import observe
langfuse = get_client()

@observe()
def judge_answer(answer: str, context: str) -> Dict[str, float]:
    """用小模型评判答案是否基于 context"""
    # 构造 judge prompt
    judge_prompt = f"""你是一个事实核查员。请判断以下答案是否严格基于提供的知识库内容。
知识库内容：
{context}

答案：
{answer}

请只回答 YES 或 NO。"""
    
    from litellm import completion
    judge_response = completion(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": judge_prompt}],
        temperature=0
    )
    is_factual = "YES" in judge_response.choices[0].message.content.upper()
    
    # 记录 judge 结果到 LangFuse
    langfuse.score_current_trace(
        name="factual_score",
        value=1.0 if is_factual else 0.0,
        data_type="NUMERIC"
    )
    return {"factual": is_factual}

# 在 generate_answer 中调用
def generate_answer(state: RAGState) -> Dict[str, Any]:
    # ... 之前的生成逻辑 ...
    answer_text = response.choices[0].message.content
    
    # 关键：在此处调用 judge
    judge_result = judge_answer(answer_text, state["context"])
    
    # 若不 factual，记录 warning
    if not judge_result["factual"]:
        langfuse.score_current_trace(
            name="hallucination_warning",
            value=1,
            data_type="NUMERIC",
            comment=f"Answer not grounded in context. Answer: {answer_text[:50]}..."
        )
    
    return {"messages": [HumanMessage(content=answer_text)]}

这样，每次 trace 在 LangFuse Dashboard 中，你会看到 factual_score 的数值分布，以及 hallucination_warning 的出现频次。结合 trace 的 input / output ，可快速定位是 retrieve 返回的 context 不足，还是 generate_answer 的 prompt 设计缺陷。

4. 避坑指南：那些只有踩过才懂的“静默陷阱”

4.1 LangChain 的 Memory 陷阱： ConversationBufferMemory 的内存泄漏

新手最爱用 ConversationBufferMemory ，认为它能记住历史。但它的 chat_memory 默认是 InMemoryChatMessageHistory ，所有消息存在 Python list 里。当用户连续提问 100 次， messages list 就有 200 条（user+ai 各一）， ConversationBufferMemory.load_memory_variables() 会把全部 200 条塞进 prompt，瞬间超 context window。更糟的是， load_memory_variables() 每次都返回新 dict，旧对象不释放，导致内存持续增长。

解决方案 ：永远用 ConversationSummaryBufferMemory 或 ConversationBufferWindowMemory 。前者用 LLM 总结历史（ llm = ChatOpenAI(model="gpt-3.5-turbo") ），后者只保留最近 k 轮（ k=5 ）。生产环境必须配置 max_token_limit=2000 ，否则 summary 本身可能超限。

实操心得：我们曾在线上环境发现内存每小时涨 50MB， tracemalloc 定位到 ConversationBufferMemory 。改用 ConversationBufferWindowMemory(k=3, max_token_limit=1500) 后，内存稳定在 120MB。记住： Buffer 不等于“缓冲”，它等于“累积”。

4.2 LlamaIndex 的 retriever 陷阱： VectorIndexRetriever 的“假相关性”

VectorIndexRetriever 默认返回 top-k 最相似 chunk，但相似度分数（score）是向量余弦值，范围 [-1,1]。很多文档 chunk 的 score 都在 0.6~0.7，用户看不出区别。更危险的是，当用户问“如何重置密码”，而知识库只有“忘记密码怎么办”，向量相似度可能高达 0.75，但答案完全错误。

解决方案 ：强制设置 similarity_top_k=3 + vector_store_query_mode="default" ，并在 query_engine 中加入 response_synthesizer 的 refine 模式：

from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.response_synthesizers import get_response_synthesizer

# 关键：使用 refine 模式，让 LLM 逐个评估每个 chunk 的相关性
response_synthesizer = get_response_synthesizer(
    response_mode="refine",  # 而非 "compact" 或 "tree_summarize"
    llm=llm
)

query_engine = RetrieverQueryEngine(
    retriever=retriever,
    response_synthesizer=response_synthesizer
)

refine 模式会让 LLM 先看第一个 chunk，判断是否相关；不相关则看第二个，依此类推。这比单纯排序可靠得多，代价是延迟增加 200ms。

4.3 LangGraph 的 stream() 陷阱： stream() 与 invoke() 的语义鸿沟

stream() 返回 generator， invoke() 返回 final state。新手常在 stream() 中试图 return ，导致 StopIteration 。更隐蔽的坑是： stream() 的 config={"callbacks": [...]} 只对 stream() 本身生效，若你在 node 函数里调用 llm.invoke() ，它不会自动继承 stream() 的 callback。

解决方案 ： stream() 仅用于 UI 流式输出， invoke() 用于后台批处理。若需在 stream() 中监控 node 内部，必须在每个 node 函数里显式传入 handler：

def generate_answer(state: RAGState, langfuse_handler=None) -> Dict[str, Any]:
    # 显式使用 handler
    if langfuse_handler:
        from langfuse.langchain import CallbackHandler
        response = llm.invoke(..., config={"callbacks": [langfuse_handler]})
    else:
        response = llm.invoke(...)
    return {...}

4.4 LangFuse 的 trace 陷阱： trace_id 重复导致数据覆盖

当多个请求并发调用 langfuse.trace() ，若未显式传入 id ，LangFuse 会自动生成 trace_id 。但在高并发下，可能生成相同 ID，导致 trace 数据互相覆盖，Dashboard 中看到“一个 trace 显示 5 个不同用户的查询”。

解决方案 ：永远显式生成唯一 trace_id ：

import uuid
from langfuse import get_client
langfuse = get_client()

# 在请求入口生成
trace_id = str(uuid.uuid4())
langfuse.trace(
    id=trace_id,
    name="rag_query",
    input={"query": user_query},
    session_id=user_session_id  # 关键：关联用户会话
)

# 后续所有 node 都用此 trace_id
config = {"callbacks": [CallbackHandler(trace_id=trace_id)]}

session_id 是另一关键，它让 LangFuse 能聚合同一用户的所有 trace，生成会话级分析报告。

4.5 LiteLLM 的 fallbacks 陷阱： fallbacks 不是“备胎”，而是“降级策略”

fallbacks=["gpt-3.5-turbo", "claude-3-haiku"] 看似简单，实则隐含风险：当 gpt-4o 超时，LiteLLM 会立即重试 gpt-3.5-turbo ，但 gpt-3.5-turbo 的 prompt 格式、temperature、max_tokens 可能与原请求不匹配，导致答案质量骤降。

解决方案 ：用 litellm.Router 精确控制 fallback：

from litellm import Router

router = Router(
    model_list=[
        {
            "model_name": "gpt-4o",
            "litellm_params": {
                "model": "gpt-4o",
                "api_key": os.getenv("OPENAI_API_KEY"),
                "timeout": 30
            }
        },
        {
            "model_name": "gpt-3.5-turbo",
            "litellm_params": {
                "model": "gpt-3.5-turbo",
                "api_key": os.getenv("OPENAI_API_KEY"),
                "timeout": 10,  # 降级模型 timeout 更短
                "temperature": 0.3  # 降级时降低 creativity
            }
        }
    ],
    fallbacks=[{"gpt-4o": ["gpt-3.5-turbo"]}]
)

# 调用时
response = router.completion(
    model="gpt-4o",  # 指定主模型
    messages=[...],
    temperature=0.7,  # 主模型参数
    max_tokens=1024
)

Router 会确保 fallback 时， temperature 、 max_tokens 等参数自动适配降级模型的特性，而非简单复制。

5. 框架选型决策树：一份可打印贴在显示器边的速查表

面对具体需求，如何快速决策？我们总结了一张决策树，覆盖 95% 的常见场景。它不追求理论完美，只回答“此刻该敲哪行代码”。

场景描述	优先框架	关键原因	替代方案	风险提示
快速验证一个想法，2小时内要能演示	LangFlow	拖拽即得 UI，导出代码可读性强，无需环境配置	LangChain + Streamlit	LangFlow 导出代码不可维护，仅限 PoC
需要处理大量 PDF/Word，且含复杂表格和公式	LlamaIndex	pymupdf4llm 支持精准表格提取， MetadataMode.ALL 保留结构信息	LangChain UnstructuredPDFLoader	Unstructured 对表格支持弱，常需额外 OCR
构建客服对话机器人，需记住用户偏好（如语言、产品型号）	LangGraph	State 天然支持长期记忆， interrupt() 可随时人工介入	LangChain ConversationSummaryBufferMemory	Summary Memory 会丢失细节，无法做条件分支
企业内网部署，所有模型必须本地运行（Ollama/vLLM）	LiteLLM + LangChain	LiteLLM 统一抽象 Ollama/vLLM/Azure，LangChain 提供成熟 Tool 生态	直接调用 Ollama API	手写适配层易出错，无熔断/重试/路由
上线后发现答案常“一本正经胡说八道”，需快速定位是检索问题还是生成问题	LangFuse	trace 可分别查看 retriever 输出和 llm.invoke() 输入，精确到 token	自研日志	自研日志难统一格式，无法做跨框架关联
团队有 Java 背景，希望复用现有 Spring Boot 微服务	LangChain (Java)	LangChain 官方提供 Java SDK，可无缝集成 Spring @Service	Python 微服务 + REST	跨语言调用增加延迟和运维复杂度
需要支持多轮追问（用户问“怎么配置”，答完后问“需要重启吗？”）	LangGraph	StateGraph 的 stream() 天然支持多轮状态延续，无需手动管理 history	LangChain AgentExecutor	AgentExecutor 的 memory 在多轮中易混乱，调试困难

这张表的核心逻辑是：**不要问“哪个框架更好”，而要问“哪个框架能让这个问题消失得最快”