个人主页 for_ever_love__ 欢迎各位大佬莅临其他栏目: 大模型开发从0到1 其他栏目: iOS项目总结大全 其他栏目: 我想学python了 其他栏目: iOS UI 文章目录LangChain 自定义 Runnable把任意 Python 逻辑接进 LCEL 链一、RunnableLambda最快的方式1.1 需要 config 时写两个参数1.2 局限性二、RunnableGenerator支持流式的自定义2.1 最简单的形式2.2 异步版本2.3 一个实际例子给流式输出加前缀/过滤2.4 什么时候需要它三、继承 Runnable完全控制3.1 用 chain 装饰器简化四、几个实用模式4.1 分支一处输入多路处理4.2 条件路由4.3 透传保留原始输入4.4 动态配置整条链五、调试看清链的结构六、常见坑七、小结LangChain 自定义 Runnable把任意 Python 逻辑接进 LCEL 链RunnableLambda大家都会用但它有个致命限制——不支持流式。想让自定义逻辑也能逐 token 输出得用RunnableGenerator。这篇讲三种自定义方式、它们的边界以及什么时候该直接继承 Runnable。一、RunnableLambda最快的方式fromlangchain_core.runnablesimportRunnableLambdadefto_upper(text:str)-str:returntext.upper()stepRunnableLambda(to_upper)chainprompt|llm|StrOutputParser()|step chain.invoke(hello)# HELLO...或者用装饰器/管道写法fromlangchain_core.runnablesimportRunnableLambdaRunnableLambdadefstep(text:str)-str:returntext.upper()# 或者直接用 | 管道LangChain 会自动包装chainprompt|llm|StrOutputParser()|to_upper1.1 需要 config 时写两个参数defstep(data:dict,config:dict)-str:userconfig[configurable].get(user_id,anon)returnf[{user}]{data[text]}chain...|RunnableLambda(step)chain.invoke({text:hi},config{configurable:{user_id:u1}})1.2 局限性RunnableLambda的输出是一次性的——函数返回什么就输出什么没法边算边吐。在流式链里它表现得像一个缓冲器chainprompt|llm|StrOutputParser()|RunnableLambda(to_upper)forchunkinchain.stream(讲个故事):print(chunk,end)# 前面的 LLM 是流式的但经过 to_upper 后……实际上普通RunnableLambda会把整个输入攒齐再处理所以你看到的是攒一大段然后一次性吐出大写版本而不是逐字变换。对短文本无所谓对长输出体验很差。二、RunnableGenerator支持流式的自定义想让自定义逻辑逐块处理并输出用生成器。2.1 最简单的形式fromlangchain_core.runnablesimportRunnableGeneratordefstreaming_upper(chunks):接收一个 chunk 迭代器yield 处理后的 chunkforchunkinchunks:yieldchunk.upper()chainprompt|llm|StrOutputParser()|RunnableGenerator(streaming_upper)forchunkinchain.stream(讲个故事):print(chunk,end,flushTrue)# 真正逐块输出关键区别RunnableGenerator的同步形式接收一个迭代器参数异步形式接收一个异步迭代器。2.2 异步版本asyncdefastreaming_upper(chunks):asyncforchunkinchunks:yieldchunk.upper()chainprompt|llm|StrOutputParser()|RunnableGenerator(astreaming_upper)awaitchain.ainvoke(...)2.3 一个实际例子给流式输出加前缀/过滤defmask_sensitive(chunks):流式输出时屏蔽手机号importrebufferforchunkinchunks:bufferchunk# 只在能确定不含手机号开头时才吐出ifnotre.search(r1\d{0,10}$,buffer):yieldbufferbuffereliflen(buffer)20:# 缓冲太长说明是手机号打码后吐出yieldre.sub(r1\d{10},1**********,buffer)bufferifbuffer:yieldbufferchainprompt|llm|StrOutputParser()|RunnableGenerator(mask_sensitive)注意这个缓冲到能判断为止的模式——流式处理里凡是涉及跨 chunk 的判断脱敏、断句、JSON 解析都需要自己做缓冲。这是流式自定义逻辑的核心难点。2.4 什么时候需要它场景用不用简单变换大小写、拼接Lambda 够了但想要真流式就用 Generator逐块脱敏 / 过滤必须 Generator给前端做 SSE 格式化必须 Generator一次性聚合如统计总长度Lambda 就行三、继承 Runnable完全控制当你的逻辑需要持有状态、需要精细控制每个方法时继承Runnablefromlangchain_core.runnablesimportRunnablefromtypingimportAny,Iterator,AsyncIteratorclassWordCounter(Runnable):统计流过的词数并在结尾附上统计def__init__(self,label:str统计):self.labellabel# 可以持有状态/配置definvoke(self,input:Any,configNone)-str:countlen(str(input).split())returnf{input}\n\n---\n{self.label}{count}个词defstream(self,input:Any,configNone)-Iterator[str]:# 逐块透传同时累加total0forchunkininput:totallen(str(chunk).split())yieldchunkyieldf\n\n---\n{self.label}{total}个词asyncdefainvoke(self,input:Any,configNone)-str:returnself.invoke(input,config)asyncdefastream(self,input:Any,configNone)-AsyncIterator[str]:total0asyncforchunkininput:totallen(str(chunk).split())yieldchunkyieldf\n\n---\n{self.label}{total}个词必须实现的方法invoke其余可选。没实现stream的话LangChain 会退化成invoke 完一次性输出。3.1 用 chain 装饰器简化fromlangchain_core.runnablesimportchainchaindefcustom(input:Any)-str:returnf处理后{input}chain_with_xprompt|llm|customchain装饰器把普通函数变成 Runnable支持同步/异步、普通函数/生成器四种组合chaindefgen(input):yield...# 同步生成器 → 支持流式chainasyncdefagen(input):yield...# 异步生成器 → 支持异步流式日常用chain就够了比手写类简洁得多。四、几个实用模式4.1 分支一处输入多路处理fromlangchain_core.runnablesimportRunnableParallel analyzeRunnableParallel({summary:summary_chain,keywords:keyword_chain,sentiment:sentiment_chain,})resultanalyze.invoke({text:long_text})# {summary: ..., keywords: [...], sentiment: positive}三路并发执行省时间。4.2 条件路由fromlangchain_core.runnablesimportRunnableBranch branchRunnableBranch((lambdax:退款inx[question],refund_chain),(lambdax:物流inx[question],logistics_chain),general_chain,# 兜底最后一个不带条件)chainbranch兜底分支必须放最后且不带条件。4.3 透传保留原始输入fromlangchain_core.runnablesimportRunnablePassthrough chainRunnableParallel({answer:rag_chain,question:RunnablePassthrough(),# 原样透传输入})resultchain.invoke(二线城市住宿标准)# {answer: ..., question: 二线城市住宿标准}4.4 动态配置整条链innerprompt|llm outerinner.with_config(run_nameqa-step,tags[prod],)with_config给某个环节绑定固定 config不用每次调用都传。五、调试看清链的结构chain.get_graph().print_ascii()会打印出链的拓扑结构复杂链非常有用------------- | PromptInput | ------------- * * * -------------- | ChatPrompt | -------------- * * * ----------------- | ChatOpenAI | -----------------还可以看每个节点的输入/输出 schemaprint(chain.input_schema.schema())print(chain.output_schema.schema())六、常见坑现象原因解法流式变一次性用了RunnableLambda换RunnableGenerator或chain 生成器拿不到 config函数只写一个参数改成(inputs, config)异步报错async generator同步函数配了异步调用异步用astream版本类型校验报错输入输出类型不匹配显式声明input_schema/output_schema链太深看不懂—chain.get_graph().print_ascii()七、小结RunnableLambda最快但不支持真流式——它会攒齐再输出RunnableGenerator支持逐块处理适合脱敏、SSE 格式化等场景跨 chunk 的判断需要自己做缓冲需要持有状态/完全控制时继承Runnable但日常用chain装饰器更简洁支持同步/异步 × 普通/生成器四种组合组合工具RunnableParallel并发分支、RunnableBranch条件路由兜底分支放最后、RunnablePassthrough透传原始输入调试用chain.get_graph().print_ascii()看拓扑。下一篇讲降级与重试——模型挂了怎么自动换路。