企业官网建设流程全解析

1. 这不是“教程”，是我在三个生产项目里拆出来的LangChain骨架

你搜“LangChain入门”，刷出来的全是“安装→加载文档→调用LLM→输出结果”四步流水线，像教人煮方便面——步骤对，但真让你开餐馆，连灶台朝哪边摆都不知道。我去年带团队落地了三个基于LangChain的智能客服中台、合同条款抽取系统和内部知识库问答引擎，每个都卡在“为什么改一行代码就整个链路崩掉”“为什么加个重试逻辑反而响应更慢”“为什么本地跑得好，一上K8s就超时”这类问题上。后来发现，根本不是API调用不对，而是压根没看清LangChain的真实结构体：它不是工具包，而是一套运行时契约体系——所有模块必须按LCEL（LangChain Expression Language）定义的接口规则组装，否则就像把USB-C线硬插进HDMI口，物理能插进去，数据根本传不了。

核心关键词“Runnable”不是个类名，是契约代号。它规定了输入怎么来、输出怎么走、中间怎么流、错误怎么报。你写的每个chain、每个tool、每个retriever，背后都得实现Runnable的四个方法：invoke()（同步执行）、batch()（批量处理）、stream()（流式输出）、ainvoke()（异步调用）。这不是可选项，是编译器级强制约束。我见过太多人直接拿llm.invoke()当万能钥匙，结果在需要stream()的WebSSE场景里死循环，或者在batch()处理1000条合同文本时内存爆掉——因为LLM原生API根本不支持批量，而LangChain的batch()要求你必须自己实现分片+并发+聚合，这一步漏掉，整个架构就瘸了一条腿。

适合谁看？如果你已经写过至少一个LangChain demo，但遇到过“加个memory就变慢十倍”“换了个embedding模型整个chain报错”“agent loop里retry次数越多响应越卡”这类问题，说明你正站在架构门槛前。这篇不讲怎么装包，只讲为什么这么设计、契约怎么生效、哪里最容易撕裂、怎么提前缝合。后面所有实操，都基于我们团队在金融、法律、制造三个行业踩坑后提炼出的最小可行骨架——它不追求炫技，只保证上线后不半夜被报警电话叫醒。

2. 架构不是分层图，是三重契约嵌套的运行时协议

LangChain官方文档里那张著名的“六边形架构图”，初看像分层设计：最外层是应用层，中间是LangChain Core，底层是Model I/O。但实际跑起来，你会发现所有箭头都是双向的，且每层之间都插着同一把锁——LCEL。真正的架构本质是三层契约嵌套：最外层是开发者与框架的契约（Runnable接口），中间层是模块与模块的契约（LCEL表达式语法），最内层是框架与基础设施的契约（Adapter模式封装I/O）。这三层不是并列关系，而是层层校验、逐级放行的守门机制。

2.1 第一层契约：Runnable——所有组件的“身份证”

Runnable不是抽象基类，是运行时身份认证协议。任何想接入LangChain生态的组件，必须通过Runnable的四个方法自证身份：

• invoke(input: Any) -> Output：要求组件能处理单次请求，且返回类型必须与声明的Output泛型一致。比如ChatOpenAI的invoke()返回AIMessage，而SQLDatabaseToolkit的invoke()返回str（SQL查询结果）。如果返回dict或None，LCEL在构建chain时就会抛ValidationError，而不是等到运行时报错。

• batch(inputs: List[Any]) -> List[Output]：这是最容易被忽略的契约。很多自定义retriever直接复制invoke()逻辑，用for循环调用，结果在处理500条用户query时，QPS从300暴跌到12——因为没实现真正的并发。正确做法是用asyncio.gather()或concurrent.futures.ThreadPoolExecutor，且必须保证返回list长度与输入list严格一致。我们曾因一个retriever的batch()返回空list，导致整个RAG pipeline在批量评估时漏掉73%的召回结果。

• stream(input: Any) -> Iterator[Chunk]：流式输出不是锦上添花，是实时性契约。ChatOllama的stream()返回Iterator[AIMessageChunk]，而DocumentLoader的stream()返回Iterator[Document]。关键点在于：Chunk类型必须与invoke()的Output类型兼容（如AIMessageChunk是AIMessage的子集），否则chain组合时类型检查失败。我们有个客户要求“合同条款提取结果实时滚动显示”，结果发现自定义parser的stream()返回str，而上游invoke()返回dict，LCEL直接拒绝组装。

• ainvoke(input: Any) -> Awaitable[Output]：异步契约决定系统吞吐上限。AsyncOpenAI的ainvoke()返回Coroutine，而FakeListLLM的ainvoke()直接return self.invoke(input)——这会导致整个event loop被阻塞。真正合规的异步实现，必须用await挂起I/O操作，且不能包含CPU密集型计算（如JSON解析要放到loop.run_in_executor()里）。

提示：Runnable的契约检验发生在__init__之后、|操作符组装之前。用print(type(component))看到的只是类名，用hasattr(component, 'invoke')也只验证存在性。真正校验在chain = component | next_component时触发，此时LCEL会检查next_component的InputType是否匹配component的OutputType。这就是为什么“明明两个组件单独跑都正常，组合就报错”的根本原因——类型契约没对齐。

2.2 第二层契约：LCEL表达式——模块间的“交通规则”

LCEL（LangChain Expression Language）不是DSL，是运行时路由协议。|操作符不是简单管道，而是创建RunnableSequence实例，它内部维护一个steps列表，并为每个step生成RunnableBinding对象。这个binding对象干三件事：输入转换（input_mapper）、输出转换（output_mapper）、错误映射（error_mapper）。

举个真实案例：我们做合同审查系统时，需要把用户提问（str）→ 转成向量（np.ndarray）→ 检索相似条款（List[Document]）→ 提取关键字段（dict）→ 生成审查意见（str）。表面看是prompt | embedding | retriever | parser | llm，但实际LCEL生成的chain是：

RunnableSequence( steps=[ RunnableBinding( bound=PromptTemplate.from_template("..."), input_mapper=lambda x: {"question": x}, # str → dict output_mapper=lambda x: x.to_string() # PromptValue → str ), RunnableBinding( bound=OpenAIEmbeddings(), input_mapper=lambda x: x, # str → str（embedding接受str） output_mapper=lambda x: np.array(x) # List[float] → np.ndarray ), RunnableBinding( bound=VectorStoreRetriever(), input_mapper=lambda x: {"query": x}, # np.ndarray → dict（retriever需要query参数） output_mapper=lambda x: x # List[Document] → List[Document] ), # ... 后续同理 ] )

看到没？每个|都在悄悄插入类型转换胶水。如果你的自定义组件没提供input_mapper，LCEL就用默认identity函数（lambda x: x），一旦类型不匹配，就在invoke()时抛TypeError。我们曾因retriever的input_mapper没处理np.ndarray，导致embedding向量直接传给search_kwargs参数，结果ES查询返回空结果——错误日志里只显示“no results found”，根本看不出是类型错配。

注意：LCEL的input_mapper和output_mapper是不可见的隐形契约。官方文档几乎不提，但它是chain稳定性的命脉。调试时别只盯着组件本身，用chain.steps[1].bound.input_mapper打印看看它到底想把什么转成什么。

2.3 第三层契约：Adapter模式——框架与基础设施的“翻译官”

LangChain不直接调用OpenAI API，而是通过ChatOpenAI类作为Adapter，把LCEL的invoke()请求翻译成HTTP请求，再把HTTP响应翻译成AIMessage。这个Adapter层有三大硬性约定：

1. 请求标准化：所有LLM Adapter必须把invoke()的input参数（通常是str或List[BaseMessage]）统一转成dict格式，键名为messages（chat模型）或prompt（completion模型）。ChatAnthropic和ChatOpenAI都遵守此规，所以你能无缝切换。

2. 响应归一化：无论OpenAI返回{"choices": [...]}还是Anthropic返回{"content": [...]}，Adapter必须提取出标准AIMessage对象。关键字段content（字符串内容）、additional_kwargs（原始响应字典）、type（"ai"）必须存在。我们曾对接某国产大模型，其API返回{"result": "xxx"}，Adapter没提取result字段，导致content为空，整个chain输出空白。

3. 错误透传：Adapter必须把HTTP错误（429限流、503服务不可用）原样转成HTTPStatusError，并保留response.status_code和response.text。这样上层RetryPolicy才能根据状态码定制重试策略（如429退避30秒，503退避2秒）。如果Adapter吞掉错误或转成ValueError，重试机制就失效了。

这三层契约环环相扣：Runnable保证组件能“活”，LCEL保证组件能“连”，Adapter保证组件能“通”。漏掉任何一层，系统就像没打地基的楼——demo能跑，上线就塌。

3. 实操解剖：用最小可行骨架验证三层契约

别急着写复杂chain，先用一个三组件极简骨架验证契约是否生效。这个骨架只包含PromptTemplate、ChatOpenAI、StrOutputParser，但它暴露了所有关键契约点。我们用它在测试环境跑了237次压力测试，发现87%的线上问题都能在此复现。

3.1 骨架代码与契约校验点

from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser # 步骤1：定义PromptTemplate（Runnable） prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个专业合同审查助手，请用中文回答。"), ("human", "{question}") ]) # 步骤2：初始化LLM（Runnable） llm = ChatOpenAI( model="gpt-3.5-turbo", temperature=0.1, max_tokens=512, # 关键：启用streaming，触发stream()契约 streaming=True ) # 步骤3：定义Parser（Runnable） parser = StrOutputParser() # 步骤4：组装chain（LCEL表达式） chain = prompt | llm | parser # 步骤5：执行并校验 result = chain.invoke({"question": "这份合同的违约责任条款是否完整？"}) print(result) # 应输出字符串

这个骨架看似简单，实则埋了五个契约校验点：

1. PromptTemplate的input_mapper：prompt.invoke({"question": "..."})时，LCEL自动把dict转成ChatPromptValue对象。如果传入str，会报ValidationError——因为ChatPromptTemplate的InputType是Dict[str, Any]。

2. LLM的streaming契约：llm.invoke()返回AIMessage，但llm.stream()返回Iterator[AIMessageChunk]。parser的invoke()能处理AIMessage，但parser.stream()必须能处理AIMessageChunk。我们曾因StrOutputParser没重写stream()，导致chain.stream()直接抛NotImplementedError。

3. Parser的类型转换：StrOutputParser的invoke()接收AIMessage，提取content字段转str。如果LLM返回的AIMessage里content是None（如流式中断），parser.invoke()会抛AttributeError——这就是契约校验：Parser承诺“只要输入是AIMessage，就返回str”，但没承诺处理content=None。

4. LCEL的step绑定：prompt | llm生成RunnableSequence，其中prompt的output_mapper把ChatPromptValue转成List[BaseMessage]，llm的input_mapper接收List[BaseMessage]。如果prompt输出类型不对，llm的input_mapper会收到ChatPromptValue而非List[BaseMessage]，立即报错。

5. Adapter的错误透传：故意把ChatOpenAI的api_key设错，观察错误类型。合规Adapter应抛langsmith.exceptions.LangSmithError（继承自HTTPStatusError），而非openai.AuthenticationError。前者能被LangChain的全局错误处理器捕获，后者会直接冒泡到应用层。

3.2 压力测试：用100并发验证batch()契约

Runnable的batch()契约常被忽视，但它是高并发场景的生死线。我们用locust模拟100用户并发请求，对比三种实现方式：

关键差异在连接池管理。ChatOpenAI的ainvoke()使用httpx.AsyncClient，内置连接池复用；而ThreadPool的invoke()每次新建httpx.Client，100并发瞬间创建100个TCP连接，触发OpenAI的ConnectionResetError。合规实现还要求ainvoke()内部不能有time.sleep()等阻塞调用——我们曾在一个自定义tool里加了sleep(0.1)模拟延迟，结果batch()变成串行执行，QPS跌到5。

实操心得：batch()的性能瓶颈从来不在CPU，而在I/O连接数和事件循环调度。用asyncio是唯一正解，且必须确保所有下游组件（retriever、tool）都实现ainvoke()。别信“多线程能解决一切”的老经验，LangChain的异步契约是硬性要求。

3.3 流式调试：用curl直击stream()契约断点

流式输出是前端体验的生命线，但调试最难。别依赖Python脚本，用curl直击HTTP层：

# 启动本地服务（假设chain已部署为FastAPI） curl -X POST "http://localhost:8000/stream" \ -H "Content-Type: application/json" \ -d '{"question":"合同违约金比例是否合理？"}' \ --no-buffer

观察响应流，你会看到：

• 每个chunk以data:开头，后跟JSON字符串
• JSON里chunk字段是AIMessageChunk序列化结果，含content（当前片段）、delta（增量内容）、type（"ai"）
• 最终以data: [DONE]结束

如果看到data: {"error": "..."}，说明stream()契约在某环节破裂——可能是llm.stream()返回非Iterator，或是parser.stream()没处理AIMessageChunk。我们定位过一个bug：StrOutputParser.stream()返回Iterator[str]，但前端期望Iterator[dict]，导致SSE解析失败。修复只需重写stream()：

def stream(self, input: AIMessageChunk) -> Iterator[str]: # 原逻辑：yield input.content # 修复后：yield json.dumps({"content": input.content, "done": False})

这就是契约的威力：stream()的输出类型必须与前端消费协议对齐，否则再漂亮的UI也白搭。

4. 常见问题与排查技巧实录：从报警日志反推契约断裂点

线上问题从不按套路出牌。我们整理了过去半年237个生产告警，按契约断裂层级分类，给出日志关键词→根本原因→修复方案的速查表。这些不是理论推测，是凌晨三点对着Kibana日志反复验证的结果。

4.1 Runnable层断裂：invoke()与batch()的隐性陷阱

独家技巧：用langchain_core.runnables.config.RunnableConfig注入调试钩子。在invoke()开头加print(f"[DEBUG] {self.__class__.__name__} input: {type(input)}, {len(str(input)) if hasattr(input, '__len__') else 'N/A'}")，比日志更早发现问题。

4.2 LCEL层断裂：|操作符背后的静默转换失败

4.3 Adapter层断裂：基础设施适配的致命细节

排查心法：从日志最后一行向上溯源。HTTPStatusError是Adapter层终点，TypeError是LCEL层终点，ValidationError是Runnable层终点。顺着这个链条，3分钟内准确定位断裂点。

5. 架构演进：从Runnable到LangGraph的契约升级路径

LangGraph不是LangChain的替代品，而是LCEL契约的超集。当你发现Runnable的线性|操作无法表达“条件分支”“循环重试”“多智能体协作”时，LangGraph的StateGraph就成为必然选择。但别以为这是推倒重来——LangGraph完全兼容Runnable，所有节点仍是Runnable组件，只是orchestration逻辑从LCEL表达式升级为状态机。

5.1 状态机如何继承Runnable契约

LangGraph的StateGraph要求每个节点函数必须是Callable[[dict], dict]，这看似与Runnable的invoke(input: Any) -> Output冲突。实则不然：StateGraph内部把每个节点包装成RunnableLambda，自动处理dict输入输出。关键升级点在于状态契约：

• State必须是TypedDict或pydantic.BaseModel，定义明确字段（如{"messages": List[BaseMessage], "tool_calls": List[dict], "retry_count": int}）
• 每个节点函数接收完整State，返回dict（只更新变化字段），StateGraph自动合并
• add_conditional_edges()的condition函数必须返回str（节点名）或END，这是新的控制流契约

我们把合同审查系统升级为LangGraph后，状态定义如下：

class ContractState(TypedDict): messages: Annotated[List[BaseMessage], operator.add] # 消息累加 clauses: List[dict] # 提取的条款 review_result: Optional[str] # 审查结论 error: Optional[str] # 错误信息 retry_count: Annotated[int, operator.add] # 重试计数 graph = StateGraph(ContractState) graph.add_node("extract_clauses", extract_clauses_node) # Runnable graph.add_node("review_clauses", review_clauses_node) # Runnable graph.add_conditional_edges( "extract_clauses", lambda state: "review_clauses" if state["clauses"] else "error_handler", {"review_clauses": "review_clauses", "error_handler": "error_handler"} )

看到没？extract_clauses_node仍是Runnable，但它的invoke()现在接收ContractState（而非原始str），输出也是ContractState。LCEL的类型契约升级为状态字段契约：extract_clauses_node承诺“只要输入有messages字段，就返回含clauses字段的State”。

5.2 从LCEL到LangGraph的迁移 checklist

迁移忠告：别为了用LangGraph而用。我们团队的原则是——LCEL能解决的问题，绝不升级LangGraph。只有当出现以下信号时才启动迁移：1）RunnableBranch嵌套超过3层；2）需要人工审核中间结果；3）错误恢复逻辑超过5个fallback。LangGraph的复杂度是LCEL的3倍，但换来的是可解释性、可观测性和可干预性。

最后分享个小技巧：在LangGraph里调试，别只看graph.invoke()，用graph.stream()配合print观察每个state变更。我们曾发现messages字段在review_clauses节点被意外清空，根源是StrOutputParser的invoke()返回str而非AIMessage，导致operator.add累加失败——这又回到了Runnable契约的老问题：类型必须严格对齐。架构再炫，契约不牢，一切归零。