Upsonic 表达式语言UEL完全指南用管道符组合 Prompt、模型与解析器的 Runnable 编排层【免费下载链接】gpt-computer-assistantBuild autonomous AI agents in Python.项目地址: https://gitcode.com/GitHub_Trending/gp/gpt-computer-assistant导读UELUpsonic Expression Language是 Upsonic 中负责编排的基础层它让提示模板、模型、解析器、普通 Python 函数、字典甚至子图都能通过管道符|自由拼接并用统一的invoke/ainvoke接口执行。本文基于当前仓库 documents/ai/explanation/uel/uel.md 与 src/upsonic/uel/ 源码完整讲解 UEL 的设计动机、全部组件、管道强制转换规则以及与models、graphv2等模块的集成方式。读完你将掌握如何用 5 行代码构建prompt | model | parser线性链如何用字典语法做并行 fan-out如何用RunnableBranch做条件路由如何用chain将任意函数变成链式步骤以及如何把任意链路渲染成 ASCII 树或 Mermaid 图。一、UEL 是什么一个刻意保持小巧的 LCEL 风格组合层upsonic.uel是 Upsonic 的Upsonic Expression Language—— 一个受 LangChain LCEL 启发的组合原语层。它的设计意图与 LCEL 一致声明式地构建链路、自动处理同步/异步分发、让纯 Python函数、字典、operator.itemgetter也能无缝参与组合。该模块刻意与任何具体模型/提供商解耦只定义四类东西一个抽象基类Runnable[Input, Output]通过重载__or__把a | b变成RunnableSequence一组覆盖 LLM 流水线结构模式的具象 Runnable 类型序列、并行、分支、lambda、透传/赋值、提示模板、输出解析器一个chain装饰器把普通 Python 函数变成 Runnable一个图视图RunnableGraph对任意链路或graphv2的StateGraph做内省并渲染为 ASCII 或 Mermaid。UEL 是高层 Upsonic 原语所依赖的管道plumbingUpsonic 类继承自效果upsonic.models.Model所有 provider 的基类Runnable[Any, Any]每个模型都可以作为prompt \| model \| parser的一环upsonic.graphv2.state_graph.StateGraphRunnable编译后的状态图是一等公民链步骤upsonic.uel.ChatPromptTemplateRunnable渲染成ModelRequest/ModelMessage列表upsonic.uel.BaseOutputParserRunnable消费ModelResponse二、文件夹布局与公共 APIUEL 刻意保持扁平共 11 个文件、约 2K 行代码无子目录src/upsonic/uel/ ├── __init__.py # 惰性公共 API 面 支持管道的 itemgetter ├── runnable.py # 抽象 Runnable[Input, Output] 管道操作符 ├── sequence.py # RunnableSequence —— 按序执行步骤链 ├── parallel.py # RunnableParallel —— 扇出返回字典结果 ├── branch.py # RunnableBranch —— 条件路由 ├── lambda_runnable.py # RunnableLambda coerce_to_runnable() 辅助 ├── passthrough.py # RunnablePassthrough(.assign) —— 输入透传 ├── prompt.py # ChatPromptTemplate —— 字符串/消息模板 ├── output_parser.py # BaseOutputParser、StrOutputParser、PydanticOutputParser ├── decorator.py # chain 装饰器函数 - Runnable └── graph.py # RunnableGraph —— ASCII / Mermaid 可视化惰性导入的公共面init.py 使用模块级__getattr__实现惰性加载from upsonic.uel import X只有在真正访问属性时才触发实际导入从而让import upsonic保持廉价。导出的名字与其来源导出名所在文件类型Runnablerunnable.py抽象基类RunnableSequencesequence.py线性链RunnableParallelparallel.py并发扇出RunnableLambdalambda_runnable.py包装函数RunnableBranchbranch.py条件路由RunnablePassthroughpassthrough.py恒等 /.assign(...)ChatPromptTemplateprompt.py提示构建器BaseOutputParseroutput_parser.py抽象解析器StrOutputParseroutput_parser.py文本提取器PydanticOutputParseroutput_parser.pyJSON → Pydanticchaindecorator.py装饰器itemgetter__init__.py合成支持管道的operator.itemgetter替代品其中itemgetter是本模块独有它返回一个RunnableLambda因此itemgetter(k) | something能直接在 UEL 链路内工作from upsonic.uel import itemgetter chain itemgetter(key) | (lambda x: fValue: {x}) chain.invoke({key: test}) # Value: test其实现init.py本质是调用原版operator.itemgetter(*items)生成 getter再用RunnableLambda(getter)包一层。三、Runnable 协议invoke/ainvoke/__or__/__ror__Runnable[Input, Output]是用户通常唯一需要子类化的抽象类定义于 runnable.py。核心契约如下class Runnable(ABC, Generic[Input, Output]): abstractmethod def invoke(self, input: Input, config: dict | None None) - Output: ... async def ainvoke(self, input, configNone): # 默认实现在线程池中运行同步 invoke loop asyncio.get_event_loop() return await loop.run_in_executor(None, self.invoke, input, config) def __or__(self, other) - Runnable: # a | b - RunnableSequence(steps[a, b]) # 若 self 已是 sequence则原地追加新步骤 # other 通过 coerce_to_runnable() 强制转换 def __ror__(self, other) - Runnable: # 支持 dict | runnable、callable | runnable几个关键设计点__or__负责强制转换所以prompt | model | (lambda r: r.text)无需用户手动包装 lambda 也能工作__ror__让左侧非 Runnable可调用对象或字典也能参与管道这正是{a: chain1, b: chain2} | next_step能工作的原因默认ainvoke回退到 executor 中的invoke任何子类只实现invoke就天然可 await。源码中 runnable.py 的实现细节__or__先对other调用coerce_to_runnable若self已经是RunnableSequence则扩展其 steps否则新建RunnableSequence(steps[self, other_runnable])。__ror__则构造RunnableSequence(steps[other_runnable, self])。Runnable ABI 一览方法签名说明invoke(input, configNone)同步执行必须实现ainvoke(input, configNone)异步执行默认为线程池中的invoke__or__(other)强制转换other并返回RunnableSequence__ror__(other)对普通字典/可调用对象的反向管道支持四、六大结构性 Runnable 详解4.1 RunnableSequence线性顺序链sequence.py 维护一个扁平的list[Runnable]invoke遍历列表把第 n 步的输出串入第 n1 步的输入class RunnableSequence(Runnable[Any, Any]): def __init__(self, steps: list[Runnable]): ... def invoke(self, input, configNone): result input for step in self.steps: result step.invoke(result, config) return result async def ainvoke(self, input, configNone): result input for step in self.steps: result await step.ainvoke(result, config) return result def __or__(self, other): # 扩展而非嵌套 return RunnableSequence(stepsself.steps [coerce_to_runnable(other)]) def get_graph(self) - RunnableGraph: ... def get_prompts(self) - list[ChatPromptTemplate]: ...注意两个细节__init__要求至少一个 step否则抛ValueError__or__返回扁平扩展的新序列而非嵌套序列。get_prompts()会递归下钻通过.steps遍历RunnableSequence/RunnableParallel收集链路上的每一个ChatPromptTemplate可用于提示词管理与检查。4.2 RunnableParallel并发扇出parallel.py 把同一个输入分发给 N 个具名 runnable把各自输出汇总为一个字典RunnableParallel(jokejoke_chain, poempoem_chain).invoke({topic: bears}) # - {joke: ModelResponse(...), poem: ModelResponse(...)}行为要点与源码一致invoke()内部调用asyncio.run(self.ainvoke(...))ainvoke()用asyncio.gather对runnable.ainvoke(input, config)做真正并行。任一分支抛异常时其余未完成任务会被取消并向上传播异常见 parallel.py每个 kwargs 值都会被强制转换Runnable保留dict变成嵌套的RunnableParallel.from_dict任何可调用对象包装为RunnableLambda其他类型抛TypeError由于coerce_to_runnable(dict)返回RunnableParallel.from_dict(dict)纯字典可直接入链chain {joke: joke_chain, poem: poem_chain} | next_step等价于RunnableParallel(joke…, poem…) | next_step。4.3 RunnableBranch条件路由分支.py 为链路提供命令式的if/elif/.../elseRunnableBranch( (lambda x: upsonic in x[topic].lower(), upsonic_chain), (lambda x: anthropic in x[topic].lower(), anthropic_chain), general_chain, # 默认分支 —— 必须是最后一个参数 )__init__中的校验规则位置必需形式失败最后一个位置参数Runnable或可调用对象不能是元组ValueError前面的位置参数(condition, runnable)二元组ValueErrorcondition可调用对象ValueError运行期行为条件按顺序求值第一个返回真值的分支胜出后续分支不再求值求值condition(input)时抛异常被当作不匹配continue而非硬失败源码见 branch.pyainvoke会额外检测inspect.iscoroutinefunction(condition)对协程条件进行await所有分支都不匹配时调用默认 runnable。4.4 RunnableLambda 与 coerce_to_runnableRunnableLambda(func)包装任意可调用对象lambda_runnable.pyclass RunnableLambda(Runnable[Input, Output]): def __init__(self, func): ... def invoke(self, input, configNone): if self.is_coroutine: # 若从异步上下文调用分发到拥有自己事件循环的线程 try: asyncio.get_running_loop() with concurrent.futures.ThreadPoolExecutor() as ex: return ex.submit(asyncio.run, self.func(input)).result() except RuntimeError: return asyncio.run(self.func(input)) return self.func(input) async def ainvoke(self, input, configNone): return await self.func(input) if self.is_coroutine \ else await asyncio.get_event_loop().run_in_executor(None, self.func, input)coerce_to_runnable(thing)是核心类型桥被__or__、RunnableSequence.__or__、RunnablePassthrough、RunnableBranch共同使用输入类型返回值Runnable原样返回dictRunnableParallel.from_dict(dict)可调用对象RunnableLambda(callable)其他TypeError4.5 RunnablePassthrough恒等与赋值passthrough.py 默认是恒等变换配合.assign(**kwargs)则向输入字典合并新键且赋值按顺序计算——后面的赋值可以读取前面已算好的值chain ( RunnablePassthrough.assign( formatted_questionlambda x: fQuestion: {x[question]}, ).assign( contextlambda x: retrieve_context(x[question]), ) | prompt | model )assign通过自定义描述符AssignDescriptor实现因此同一个标识符既可用作类方法RunnablePassthrough.assign(...)返回新实例也可用作实例方法existing.assign(...)合并其 assignments 并返回新实例。输入规则无赋值 → 原样返回input任意类型有赋值 →input必须是 dict否则抛TypeError每个赋值值都会coerce_to_runnable然后用运行中的结果字典调用返回值存入对应键。4.6 ChatPromptTemplate提示模板prompt.py 构建两种产物普通格式化字符串旧式template构造器或ModelRequest/ModelMessage列表多消息提示。两个工厂方法ChatPromptTemplate.from_template(Tell me a {adj} joke about {topic}) # - 单条 human 消息模板变量通过正则自动提取 ChatPromptTemplate.from_messages([ (system, You are a helpful assistant), (placeholder, {variable_name: chat_history}), (human, Tell me about {topic}), ])支持的消息角色角色归一化为systemSystemPromptPart放在第一个ModelRequest中human/userUserPromptPartai/assistantModelResponse(parts[TextPart])用于 few-shot 示例placeholder从input[variable_name]动态注入invoke强制的转换规则恰好一个SystemPromptPart且只挂到第一个ModelRequest每个ModelRequest一个UserPromptPart相邻 human 消息用空格合并每个ModelResponse一个TextPart针对aifew-shot 轮会话按 Request → Response → Request → Response 交替若input已经是ModelMessage或ModelMessage列表通过.kind in (request, response)检测原样返回——这让 memory/history 可以不经改动地流过提示步骤缺少必需变量抛KeyError非 dict / 非消息输入抛TypeError。Placeholder 接受(role, content)元组列表视为聊天历史或字符串作为用户提示追加。源码中从upsonic.messages导入ModelRequest、SystemPromptPart、UserPromptPart、ModelResponse并用RequestUsage与now_utc()构造 few-shot AI 轮的占位ModelResponse。4.7 输出解析器家族从 ModelResponse 到 str / Pydanticoutput_parser.py 中BaseOutputParser[T]是Runnable[ModelResponse, T]子类只需实现parse(response: ModelResponse) - T类parse行为StrOutputParser若最后一个 part 是TextPart返回response.parts[-1].content否则返回PydanticOutputParser(Model)对最后一个TextPart做 JSON 解析并用传入的 Pydantic 类校验优先model_validate否则Model(**parsed)空 parts、非文本末 part、JSON 错误或 schema 校验失败均抛ValueError两者都是prompt | model | parser链路的现成尾步。五、chain装饰器函数即 Runnabledecorator.py 把同步或异步函数变成 Runnablechain def custom_chain(text): p ChatPromptTemplate.from_template(Tell me a joke about {topic}) out infer_model(openai/gpt-4o).invoke(p.invoke({topic: text})) return (ChatPromptTemplate.from_template(Subject of joke: {joke}) | infer_model(openai/gpt-4o)).invoke({joke: out}) custom_chain.invoke(bears)值得注意的语义内部构建一个私有ChainRunnable(Runnable)并调用functools.update_wrapper因此__name__、__doc__等元数据得以保留若被包装函数返回一个Runnable装饰器会用同样的输入和同样的config调用那个返回的 runnable从而支持动态链路chain def dynamic(input_): return complex_chain if input_[use_complex] else simple_chain同步invoke携带async函数体时若在运行中的事件循环内调用会快速失败——抛RuntimeError引导调用方改用await runnable.ainvoke(...)源码见 decorator.py。六、RunnableGraph链路内省与可视化graph.py 提供静态内省与可视化能力方法输出to_ascii()/print_ascii()带|/v连接符与并行分支标记的缩进 ASCII 树to_mermaid()graph TD ...Mermaid 源码并行分叉用带分支键标签的虚线边顺序边用get_structure_details()逐节点拆解并行分支、合并目标、顺序边它接受 UELRunnable通过_build_graph自动展开RunnableSequence/RunnableParallel或graphv2.StateGraph/CompiledStateGraph通过_build_state_graph普通StateGraph边变成顺序edges_to条件边变成parallel_branches渲染为虚线 Mermaid 箭头让动态路由在视觉上可区分引用到START__start__和END时添加对应标记若某节点同时有普通边与条件边只保留条件视图运行期条件优先。内部的_StateGraphNodeRef是一个不可执行的Runnable代理纯粹作为标签载体用于可视化那些节点本身不是 runnable 的图。七、管道强制转换规则速查表左侧右侧|之后的类型RunnableRunnableRunnableSequenceRunnable可调用对象RunnableSequence右端包装为RunnableLambdaRunnabledict[str, Runnable|callable]RunnableSequence右端变为RunnableParalleldictRunnableRunnableSequence左端变为RunnableParallel可调用对象RunnableRunnableSequence左端包装为RunnableLambda八、与 Upsonic 其余模块的集成UEL 是基础层多个模块依赖它模块如何使用 UELsrc/upsonic/models/init.py基类Model是class Model(Runnable[Any, Any])重写invoke/ainvoke因此任何 provider 子类OpenAI、Anthropic、Bedrock…都能放进 UEL 链路模型模块还从upsonic.uel.output_parser导入StrOutputParser、PydanticOutputParser用于结构化输出配置src/upsonic/graphv2/state_graph.pyStateGraph及CompiledStateGraph继承upsonic.uel.runnable.Runnable状态图本身就是链步骤图模块也调用upsonic.uel.graph的RunnableGraph渲染节点/边结构src/upsonic/messages被消费UEL 从upsonic.messages导入ModelRequest、ModelResponse、SystemPromptPart、UserPromptPart、TextPart用于prompt.py与output_parser.pysrc/upsonic/usage、src/upsonic/_utilsprompt.py导入RequestUsage与now_utc()构造 few-shot AI 轮的占位ModelResponse因此典型的 Upsonic Agent 路径是ChatPromptTemplate (Runnable) | Model (Runnable, provider 子类) | OutputParser (Runnable)同样的组合方式也用于graphv2节点内部以及更高层的预构建 Agentsrc/upsonic/prebuilt/...中任何需要确定性 prompt-model-parse 步骤的场景。九、端到端示例9.1 线性流水线追踪from upsonic.uel import ChatPromptTemplate, StrOutputParser from upsonic import infer_model prompt ChatPromptTemplate.from_template(Tell me a joke about {topic}) model infer_model(openai/gpt-4o) parser StrOutputParser() chain prompt | model | parser result chain.invoke({topic: bears})运行期发生了什么prompt | model——Runnable.__or__执行coerce_to_runnable(model)→ 返回model本身已是Runnable构建RunnableSequence(steps[prompt, model])。(prompt | model) | parser——RunnableSequence.__or__执行返回RunnableSequence(steps[prompt, model, parser])扁平非嵌套。chain.invoke({topic: bears})prompt.invoke({topic: bears})→ModelRequest(parts[UserPromptPart(Tell me a joke about bears)])model.invoke(ModelRequest)→ 调用 provider返回ModelResponseparser.invoke(ModelResponse)→ 返回response.parts[-1].content作为str。9.2 并行 赋值 分支的组合链路from upsonic.uel import ( ChatPromptTemplate, RunnablePassthrough, RunnableBranch, RunnableParallel, StrOutputParser, itemgetter, ) from upsonic import infer_model model infer_model(openai/gpt-4o) summarize ChatPromptTemplate.from_template(Summarize: {text}) | model | StrOutputParser() classify ChatPromptTemplate.from_template(Classify topic of: {text}) | model | StrOutputParser() short_chain ChatPromptTemplate.from_template(Short reply about: {text}) | model | StrOutputParser() long_chain ChatPromptTemplate.from_template(Long essay about: {text}) | model | StrOutputParser() pipeline ( RunnablePassthrough.assign( word_countlambda x: len(x[text].split()), ) | RunnableParallel( summarysummarize, topicclassify, originalitemgetter(text), ) | RunnableBranch( (lambda d: len(d[original]) 200, short_chain | (lambda s: {reply: s})), long_chain | (lambda s: {reply: s}), ) ) print(pipeline.invoke({text: Once upon a time ...}))逐步解读RunnablePassthrough.assign(word_count...)—— 复制输入字典基于运行中字典计算并存储word_countRunnableParallel(summary…, topic…, originalitemgetter(text))—— 通过asyncio.gather并发执行 3 个ainvoke返回{summary: str, topic: str, original: str}RunnableBranch—— 检查len(d[original]) 200路由到short_chain或默认的long_chain此处分支臂本身是包含尾部 lambda把字符串包进 dict的RunnableSequence。9.3 可视化链路print(pipeline.get_graph().to_ascii())输出缩进树并行分支以├─/└─标记并带并行键名。pipeline.get_graph().to_mermaid()返回可粘贴进任意 Mermaid 渲染器的源码并行分叉以带分支键标签的虚线箭头输出顺序边为。9.4 可视化 StateGraphfrom upsonic.graphv2.state_graph import StateGraph g StateGraph(SomeState).add_node(a, a_fn).add_node(b, b_fn).add_edge(__start__, a).add_edge(a, b) print(g.get_graph().to_mermaid()) # 内部使用 RunnableGraphStateGraph的条件边渲染为虚线 Mermaid 箭头与静态边视觉区分引用到时添加START/END。9.5 通过 chain 构建动态链路chain def smart_route(data: dict): if data[needs_rag]: return rag_chain # ChainRunnable 会用 data 调用它 return ChatPromptTemplate.from_template(Answer: {q}) | model | StrOutputParser() smart_route.invoke({needs_rag: True, q: What is Upsonic?})smart_route本身就是一个Runnable因此可以作为任何更大链路中的合法步骤。十、测试验证与实现佐证仓库中针对 UEL 有直接的测试覆盖tests/smoke_tests/uel/test_all_scenarios.py 从upsonic.uel导入ChatPromptTemplate、RunnablePassthrough、StrOutputParser覆盖四种内存模式auto / always / never / record_all、placeholder 注入、多链 RAG 模式、空历史、顺序占位符调用等场景——验证了ChatPromptTemplate的 placeholder 动态注入与消息透传行为正是 UEL 提示模板与模型内存系统协同工作的关键路径。另有 tests/unit_tests/test_lcel.py 与 tests/unit_tests/test_lcel_focused_issues.py 从 LCEL 视角验证链式组合的正确性。结语UEL 刻意保持小巧它只形式化步骤如何连接Runnable、管道操作符、sequence/parallel/branch/passthrough、任意 Python 如何进入链路RunnableLambda、coerce_to_runnable、chain以及链路如何被内省RunnableGraph。一切领域相关的部分模型、工具、内存、Agent、RAG、安全都位于该文件夹之外仅仅是使用Runnable协议而已。理解 UEL就等于理解了整个 Upsonic 一切组合能力的底层语法。【免费下载链接】gpt-computer-assistantBuild autonomous AI agents in Python.项目地址: https://gitcode.com/GitHub_Trending/gp/gpt-computer-assistant创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考