Agent Zero 的 call_llm.py 源码解析基于 LangChain 的少样本流式 LLM 调用封装【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zerohelpers/call_llm.py是 Agent Zero 框架中一个轻量而关键的底层辅助模块它以约 70 行代码封装了 LiteLLM/模型调用的通用路径通过 LangChain 提示词模板完成少样本few-shot示例注入借助 LCEL 管道与astream()实现逐块流式输出并向外暴露逐块回调钩子。本文以该模块的 DOX 文档helpers/call_llm.py.dox.md为骨架结合其源码实现与框架内调用方完整讲解它的 API 契约、内部工作流程、扩展钩子集成方式以及维护约定读者读完可直接理解该封装的使用方式并掌握在 Agent Zero 中复用其能力的正确姿势。模块定位框架 LLM 调用的轻量层Agent Zero 采用扁平化的helpers/目录组织可复用框架 APIcall_llm.py与其它辅助模块一样通过同名 DOX 文档call_llm.py.dox.md记录职责与契约。DOX 文档明确给出了该模块的两层定位call_llm.py拥有运行时实现它是真正可执行、可被导入的代码call_llm.py.dox.md拥有持久化说明记录该实现的责任、契约、副作用与验证方式二者必须保持同步因为该目录刻意保持扁平。DOX 的 Purpose 段落将模块的核心价值概括为wraps LiteLLM/model calls with examples, streaming, and extension hooks——即包装模型调用并同时提供三个能力维度能力维度含义对应实现examples少样本示例注入FewShotChatMessagePromptTemplate组装示例对streaming流式输出LCEL 管道的chain.astream({})extension hooks扩展点钩子模块对外暴露的callback参数以及框架层面与extension.extensible扩展系统的配合从源码结构看该模块仅依赖langchain.prompts、langchain.schema、langchain_core.language_models、langchain_core.messages与typing见 helpers/call_llm.py是一个不依赖框架内部状态的纯函数式封装这正是它能够作为通用底层 API 被复用的前提。API 契约Example 类型与 call_llm 签名Example少样本示例的类型约束模块用TypedDict定义了少样本示例的数据形状class Example(TypedDict): input: str output: str每个示例由input与output两个字符串字段构成语义即给定输入期望模型给出该输出。TypedDict的使用意味着在类型检查器如 mypy下传入不符合形状的示例会得到静态提示而运行时则无需额外校验——这保持了封装层的轻量。call_llm核心异步函数DOX 文档记录的函数签名如下async call_llm(system: str, model: BaseChatModel | BaseLLM, message: str, examples: list[Example] [], callback: Callable[[str], None] | None ...)各参数的实际语义结合 helpers/call_llm.py 可归纳为参数类型说明默认值systemstr系统提示词注入最终提示词的SystemMessage必填modelBaseChatModel \| BaseLLM任意符合 LangChain 核心接口的聊天模型或 LLM 实例含 Agent Zero 自研模型包装必填messagestr用户的当前输入注入最终提示词的HumanMessage必填exampleslist[Example]少样本示例列表可为空[]callbackCallable[[str], None] \| None每收到一个流式文本块时被同步调用参数为该块文本None返回值是一个str所有流式块按序拼接后的完整响应文本。该函数是async的调用方必须使用await或在asyncio.gather等环境中并发执行。内部实现剖析从提示词组装到流式输出整个调用流程在 helpers/call_llm.py 中分为四个阶段DOX 的 Key Concepts 段落对此逐一点名ChatPromptTemplate.from_messages、FewShotChatMessagePromptTemplate、few_shot_prompt.format、chain.astream、HumanMessage、AIMessage、SystemMessage、callback。阶段一构建少样本示例模板example_prompt ChatPromptTemplate.from_messages( [ HumanMessage(content{input}), AIMessage(content{output}), ] ) few_shot_prompt FewShotChatMessagePromptTemplate( example_promptexample_prompt, examplesexamples, input_variables[], )每个少样本示例被渲染为一轮人机对话HumanMessage({input})后紧跟AIMessage({output})。FewShotChatMessagePromptTemplate会把examples列表中的每个示例都按该模式展开最终形成多轮示例对话块。值得注意的是input_variables[]——示例内容完全由examples参数静态注入不再依赖模板变量插值。随后调用few_shot_prompt.format()。该调用的返回值在代码中并未被使用其真实作用是触发模板的即时求值与校验若示例结构非法如缺少字段错误会在此处立即暴露而不是拖延到模型调用阶段起到了提前失败fail fast的作用。阶段二组装最终提示词final_prompt ChatPromptTemplate.from_messages( [ SystemMessage(contentsystem), few_shot_prompt, HumanMessage(contentmessage), ] )最终提示词按固定次序拼接三部分系统指令SystemMessage→ 少样本示例对话few_shot_prompt→ 当前用户输入HumanMessage。这一顺序符合常见的大模型提示词工程惯例先给出行为约束再用示例演示期望的输入输出形态最后交付真实任务。阶段三LCEL 管道与流式迭代chain final_prompt | model response async for chunk in chain.astream({}):final_prompt | model是 LangChain Expression LanguageLCEL的管道语法将提示词模板与模型组合成一个可调用的 Runnable 链。chain.astream({})以流式方式逐块产出模型输出——这正是流式能力的核心实现在完整响应返回之前调用方即可开始消费内容。阶段四块归一化与回调分发if isinstance(chunk, str): content chunk elif hasattr(chunk, content): content str(chunk.content) else: content str(chunk) if callback: callback(content) response content由于不同模型后端返回的块类型并不统一可能是裸字符串、带.content属性的消息对象或其它类型代码对 chunk 做了三级归一化处理直接是str→ 原样使用具有content属性如AIMessageChunk→ 取属性值并转字符串其它类型 → 兜底str()转换。每个归一化后的文本块会先交给callback若提供再累加到response。整个函数结束时返回拼接完成的完整响应。此外源码在流式循环内保留了一行被注释的# await self.handle_intervention() # wait for intervention and handle it, if paused从注释语义看这里预留了暂停时等待人工干预的挂载点说明该封装在设计上考虑了与 Agent 生命周期控制如 pause/intervention 机制的衔接空间。与框架上层调用的关系utility model 路径DOX 文档指出该模块Observed side-effect areas: model calls可观察的副作用区域模型调用并明确其属于框架公共 API调用方包括核心代码与插件。从仓库现状看helpers/history.py 在模块级通过from helpers import messages, tokens, settings, call_llm导入了该模块可见它被框架的历史/上下文子系统作为底层依赖引入。同时需要说明当前更常用的上层路径Agent 类的call_utility_model与call_chat_model见 agent.py均被标记为extension.extensible实际执行时通过模型的unified_call方法完成调用并在调用前后触发util_model_call_before/util_model_call_after等扩展钩子helpers/history.py 中的话题摘要Topic.summarize_messages正是经由agent.call_utility_model走这条上层路径。可以推断call_llm.py所提供的Example system message callback参数模型是这些上层封装在提示词组装 流式回调语义上的通用基础形态适合在需要直接控制 LangChain 模型实例、快速做少样本实验或构建独立工具时复用。扩展钩子回调与扩展系统的边界DOX 中extension hooks并非指call_llm.py内部存在call_extensions_async之类的调用该模块源码中并没有而是指两层含义回调层钩子callback参数允许调用方在流式产出每一块文本时立即处理如转发给前端、写入日志、驱动 UI 动画这是模块对外最直接的扩展点框架扩展层配合Agent Zero 的扩展系统extension.extensible、call_extensions_before/after系列见 agent.py 与 helpers/extension.py可在更高层级对模型调用进行拦截、改写与观测call_llm.py的轻量定位使其既可以独立使用也可以被上层扩展路径所替代或包裹。运行时契约与维护约定DOX 的 Runtime Contracts 与 Work Guidance 段落规定了使用与修改该模块时必须遵守的规则这同样是理解其为什么长这样的关键保留公共调用方辅助模块拥有可被核心代码与插件复用的框架 API除非所有调用方、测试与文档同步更新否则不得破坏公共函数与类的签名同步更新 DOX只要公共函数、类、持久化行为、路径/安全假设、副作用或跨模块契约发生变化就必须同步更新call_llm.py.dox.md副作用需显式且有界路径、认证、密钥、持久化、网络与子进程行为必须保持显式且受控——而call_llm.py的副作用面被刻意收敛在模型调用这一单一区域不触碰文件系统与网络配置按需内聚只有当某行为被多个模块复用且具备内聚性时才建议在helpers/下新增此类辅助函数避免无意义膨胀。验证与测试DOX 的 Verification 段落给出的验证指引是针对变更的辅助行为运行定向测试并针对涉及认证、文件系统、WebSocket、隧道、上传或密钥处理的辅助模块运行安全回归测试。同时 DOX 明确记载按名称搜索未找到直接引用call_llm的测试仓库内的测试目录tests/与各插件tests/下均无对该模块的直接单测因此建议选择最接近的行为测试或进行一次聚焦的冒烟检查。这一点在仓库中得到印证对call_llm的检索结果仅出现在 helpers/call_llm.py 自身、其 DOX 文档以及 helpers/history.py 的导入语句中没有独立测试文件。这提示开发者在修改该模块后应通过其间接调用链如历史摘要、话题压缩等依赖模型调用的路径进行冒烟验证并考虑为其补充针对性测试。小结helpers/call_llm.py以极小的代码面完成了 Agent Zero 中一次带少样本示例、支持流式输出、可逐块回调的标准模型调用封装Example定义示例数据形状call_llm通过ChatPromptTemplate与FewShotChatMessagePromptTemplate组装提示词借 LCEL 管道与astream()实现流式迭代再以统一的 chunk 归一化逻辑分发回调并聚合完整响应。DOX 文档则为该模块划定了明确的职责边界、同步维护要求与验证策略。理解这一层封装是深入 Agent Zero 模型调用链路unified_call、扩展钩子、上下文压缩的起点也为在插件或自定义工具中快速复用框架 LLM 调用能力提供了现成范式。延伸阅读call_llm.py 源码实现完整源码69 行call_llm.py DOX 文档模块级 DOX 档案history.py导入call_llm的框架调用方含话题/批量摘要等模型调用路径agent.pyget_utility_model/call_utility_model等上层可扩展模型调用接口models.py框架自研模型包装与流式解析实现extension.pyextension.extensible扩展系统实现【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zero创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考