跳到主要内容

LLM 抽象层:文本里解析工具调用、多 provider 插件化与 token 治理

30 秒导读: 这一层要解决两件事——(1) 让 MiroFlow 能接上任意一家大模型(Claude / GPT / DeepSeek / 自研 MiroThinker),(2) 把模型"说的话"变成 agent 真能执行的工具调用。MiroFlow 的关键选择是:不走各家原生的 tool_calls 接口,而是教模型在回复正文里写一段 <use_mcp_tool> XML,再用正则 + 一堆 JSON 修复函数把它抠出来。本章讲这条"把文本变成动作"的暗线,以及配套的 provider 插件化与 token 治理。

本章只讲模型侧与解析。工具怎么被真正执行、MCP server 池怎么容错,见 工具层;一次任务从头到尾怎么循环,见 编排主循环;sub-agent 怎么被当成工具调,见 分层 sub-agent。

1. 这是什么(零基础也能懂)​

一句话定义: LLM 抽象层是 MiroFlow 里所有"和大模型对话"的统一门面——不管底下是 Anthropic、OpenAI 还是别的家,上层 agent 都只调同一组方法。

它要挡住的麻烦: 每家模型 API 长得都不一样。

差异点AnthropicOpenAI Chat/Responses自研/OpenRouter
返回结构response.content 里一串 blockresponse.choices[0].message各不相同
工具调用native tool_use blocknative tool_calls / function_call有的支持有的不支持
usage 字段cache_creation/read_input_tokensprompt_tokens_details五花八门
缓存cache_control: ephemeral自动不一定有

如果上层每接一家就写一套 if-else,循环代码会被撑爆。抽象层的活就是:把这些差异全部关进一个个 provider 类里,对外只暴露"给我发消息、把响应处理成文本、从里面抠出工具调用"这几个动词。

一句话直觉: 把它想成一个"翻译插座"——上层 agent 说一种"普通话"(发消息 / 处理响应 / 抠工具调用),每个 provider 类是一个"方言转换头",负责把普通话翻成某家 API 的方言,再把方言的回复翻回普通话。

最不显然、也最值得学的一点: MiroFlow 明明各家都有 native 工具调用接口,它却偏偏不用(至少对 Claude 是这样),而是让模型把工具调用写进回复的正文文本,像这样:

我先查一下这个 PDF 里有没有编号引用:

<use_mcp_tool>
<server_name>tool-code</server_name>
<tool_name>run_command</tool_name>
<arguments>
{
"sandbox_id": "i86ayus8ryxxtaifen3bg",
"command": "pdfgrep -i '\\[[0-9]\\]' /home/user/paper.pdf"
}
</arguments>
</use_mcp_tool>

然后 MiroFlow 用正则把这段 XML 抠出来。为什么要放着现成的结构化接口不用、去解析文本?下一节回答。

2. 顶层全景(它大概怎么转)​

怎么读这张图: 从上到下是"一次模型调用"的生命周期。左边是统一门面 LLMProviderClientBase,右边是每个 provider 子类各自实现的"方言"部分。中间那条粗线(文本 → 正则 → JSON 修复)是本章的主角。

          上层编排循环(01 章)
                 │  create_message(system, history, tools, keep_tool_result)
                 ▼
        ┌───────────────────────────────────────────┐
        │  LLMProviderClientBase(抽象基类·统一门面)   │
        │  · create_message:裁历史 → 调子类 → 累计 usage │
        │  · usage 累计 / get_usage_log               │
        └───────────────┬───────────────────────────┘
                        │ 委托给具体 provider 子类
                        ▼
     ┌──────────── provider 子类(方言层)────────────┐
     │  _create_message   → 拼本家参数、发 API 请求     │
     │  process_llm_response → 把响应变成一段文本      │
     │  extract_tool_calls_info → 从文本里抠工具调用   │
     └───────────────┬───────────────────────────────┘
                     │ 把"模型说的话"交给解析器
                     ▼
        ┌───────────────────────────────────────────┐
        │  parse_llm_response_for_tool_calls          │
        │  三种入参形态 → 统一成 [{server,tool,args}]  │
        │       ├─ dict  → OpenAI Responses function_call│
        │       ├─ list  → OpenAI/DeepSeek tool_calls    │
        │       └─ str   → <use_mcp_tool> XML(正则)     │
        └───────────────┬───────────────────────────┘
                        │ args 是模型手写的脏 JSON
                        ▼
        ┌───────────────────────────────────────────┐
        │  robust_json_loads + 一票修复函数            │
        │  截断补引号 → json → json5 → 逐字段转义      │
        └───────────────────────────────────────────┘
                        │ 干净的 tool_calls
                        ▼
                 交回编排循环去执行(03 章)

部件一句话职责:

部件干什么在哪
LLMClient工厂:按配置字符串动态 import 出正确的 provider 类src/llm/client.py:13
providers/__init__.py启动时自动扫描目录、把所有 provider 类收集到命名空间src/llm/providers/__init__.py
LLMProviderClientBase抽象基类:定义统一方法 + 通用的裁历史/usage 逻辑src/llm/provider_client_base.py:27
provider 子类每家 API 一个类,实现 _create_message 等抽象方法src/llm/providers/*_client.py
parse_llm_response_for_tool_calls把三种形态的"工具调用"统一抠成 dict 列表src/utils/parsing_utils.py:584
JSON 修复族容忍模型输出的坏 JSONsrc/utils/parsing_utils.py:18-582

主线走一遍(高层): 上层调 create_message → 基类先裁历史再委托子类发请求 → 子类把响应变成一段文本 → extract_tool_calls_info 把文本喂给解析器 → 解析器正则抠出 XML、JSON 修复族把脏 args 修好 → 返回一串 {server_name, tool_name, arguments} → 交回循环去执行。

3. 核心原理(逐个机制,由浅入深)​

3.1 provider 插件化:一个字符串换一个大模型​

它要解决的小问题: 想换模型时,最好只改配置里一行,而不是改代码。

思路: 配置里写一个 provider_class 字符串(比如 "ClaudeAnthropicClient"),运行时用它去动态 import 出对应的类。这样新增一家模型 = 新写一个类文件,工厂代码一行都不用动。

真实实现: 工厂 LLMClient 拿到字符串后动态导入(src/llm/client.py:44-59):

# src/llm/client.py:44 —— 摘录
if not isinstance(provider_class, str) or not provider_class.isidentifier():
    raise ValueError(f"Invalid provider_class: {provider_class}")
providers_module = importlib.import_module("src.llm.providers")
ProviderClass = getattr(providers_module, provider_class)  # 按名字取类
client_instance = ProviderClass(task_id=task_id, task_log=task_log, cfg=config)

关键细节——为什么 getattr(providers_module, "ClaudeAnthropicClient") 能取到类? 因为 providers/__init__.py 在被 import 时,会自动遍历目录下所有 .py,把每个模块里定义的类塞进包的命名空间(src/llm/providers/__init__.py:15-26):

# src/llm/providers/__init__.py:15 —— 摘录
for module_info in pkgutil.iter_modules([package_dir]):
    module = importlib.import_module(f"{__name__}.{module_info.name}")
    for name, obj in inspect.getmembers(module, inspect.isclass):
        if obj.__module__ == module.__name__:   # 只收本模块定义的类,不收 import 进来的
            globals()[name] = obj                # 挂到包命名空间
            __all__.append(name)

obj.__module__ == module.__name__ 这个判断很重要:它保证只收本文件里定义的类,把从别处 import 进来的类(比如基类)排除掉,避免命名污染。

由此可得的扩展方式: 往 providers/ 丢一个新 xxx_client.py,里面写个继承 LLMProviderClientBase 的类,配置里把 provider_class 指向它——就接好了。目前已有的子类:ClaudeAnthropicClient、claude_openrouter、deepseek_openrouter、gpt5_openai、gpt_openai、mirothinker_sglang。

3.2 抽象基类:哪些必须各家自己实现,哪些大家共用​

它要解决的小问题: 既想让各 provider 灵活,又不想每家把通用逻辑(裁历史、累计 token)重写一遍。

思路: 用抽象基类划一条线——"方言相关的"设成 @abstractmethod 强制子类实现,"通用的"在基类里写好共用。

必须子类实现的抽象方法(src/llm/provider_client_base.py):

抽象方法职责行
_create_client造出本家 SDK 的 client 对象:85
_create_message拼本家参数、真正发一次 API 请求:90
process_llm_response把响应对象变成 (文本, 是否该停),并把 assistant 消息写回历史:101
extract_tool_calls_info从响应里抠出工具调用:108
update_message_history把工具结果按本家格式塞回历史:324
handle_max_turns_reached_summary_prompt到达最大轮数时怎么拼收尾 prompt:333

基类里共用、子类不用重写的:

  • create_message(:177)——统一入口:先 _filter_message_history 裁历史,再委托子类 _create_message,最后统一累计 usage。
  • usage 累计的四个计数器(total_input_tokens 等,:38-41)+ get_usage_log(:368)。
  • _remove_tool_result_from_messages(:115)——token 治理的核心裁剪逻辑(见 3.5)。

一个巧妙的兜底: _extract_usage_from_response(:339)在基类里给的是 OpenAI Chat Completions 格式的默认解析;Anthropic 的 usage 字段名不一样(cache_creation_input_tokens 等),所以 ClaudeAnthropicClient 重写了它(claude_anthropic_client.py:188)。"基类给默认、子类按需覆盖"——只有真不一样的家才付重写的代价。

3.3 关键设计:工具调用走文本 XML,而不是 native tool_calls​

它要解决的小问题: 让"模型想调哪个工具"这件事,在所有 provider 上用同一套逻辑处理——哪怕某些模型/接入方式根本不支持 native tool 调用。

思路(这是全章的暗线): 与其依赖每家 API 各异、且不一定都有的 native 工具调用,不如在 system prompt 里教模型:把工具调用写成一段固定格式的 <use_mcp_tool> XML,放在回复末尾。这样无论底下是谁,上层只需要一件事:从模型吐出的文本里用正则找这段 XML。

prompt 是怎么教的? 主 agent 的 system prompt 明确规定了格式与放置规则(config/agent_prompts/main_agent_prompt_gaia.py:28-50):

Tool-use is formatted using XML-style tags ...

  • Tool-use must be placed at the end of your response, top-level ...
  • The output is not expected to be valid XML and is parsed with regular expressions.

注意最后一句:它直接告诉模型"我是用正则解析的,不是严格 XML"——这为后面的容错解析定了调子。sub_worker.py、main_agent_prompt_deepseek.py、main_boxed_answer.py 里也都教了同一格式。

为什么值得这么做? 好处有几个:

  • 统一 —— 不管 Claude / GPT / 自研模型,抠工具调用的代码是同一份,不必为每家的 native 格式各写一套。
  • 可兼容不支持 tool 调用的模型 —— 只要模型会按格式写文本就行,连 native 接口都不需要。
  • 可观测/可调试 —— 工具调用就写在回复正文里,日志里一眼能看到模型"想干嘛"。

代价是:要自己解析,还要容忍模型把 XML 和 JSON 写歪——于是有了 3.4 的一大票修复函数。

三种形态,一个解析器: 虽然主打文本 XML,parse_llm_response_for_tool_calls(parsing_utils.py:584)其实同时兼容三种入参,靠类型分派:

入参类型来自处理分支
dictOpenAI Responses API 的 output[].function_callparsing_utils.py:591-634
listOpenAI/DeepSeek Chat 的 message.tool_callsparsing_utils.py:637-682
strClaude / GPT 等把响应转成的文本里的 <use_mcp_tool>parsing_utils.py:684-802

举例:DeepSeekOpenRouterClient 在响应带 native tool_calls 时,直接把 message.tool_calls(一个 list)喂进来(deepseek_openrouter_client.py:309-311),走 list 分支;而 ClaudeAnthropicClient 与 Gpt5OpenAIClient 把响应转成文本再喂,走 str/XML 分支(claude_anthropic_client.py:151-156、gpt5_openai_client.py:267-272)。三条路最后都吐出统一结构 {server_name, tool_name, arguments, id}。

XML 分支的正则(parsing_utils.py:689)——注意它对空白、大小写、标签属性都很宽松:

# parsing_utils.py:689 —— 摘录
tool_call_patterns = re.findall(
    r"<use_mcp_tool[^>]*?>\s*<server_name[^>]*?>(.*?)</server_name>"
    r"\s*<tool_name[^>]*?>(.*?)</tool_name>"
    r"\s*<arguments[^>]*?>\s*([\s\S]*?)\s*</arguments>\s*</use_mcp_tool>",
    llm_response_content_text, re.DOTALL | re.IGNORECASE,
)

server_name 与 tool_name 之所以用 - 连接又能拆开(name.rsplit("-", maxsplit=1),parsing_utils.py:596/641),是因为工具在注册时就是 f"{server['name']}-{tool['name']}" 这么拼的(见基类 convert_tool_definition_to_tool_call,provider_client_base.py:221-236)。

坏调用不静默丢弃: 解析器还专门找"没闭合的标签"记成 bad_tool_calls(parsing_utils.py:697-742),并对最常见的"漏了 </arguments>"做一次自动补齐后递归重解析(parsing_utils.py:778-800)——这样模型偶尔写漏一个闭合标签也能救回来。

3.4 容忍模型的坏 JSON:一整套修复流水线​

它要解决的小问题: 模型手写的 arguments JSON 经常不合法——引号没转义、字符串没收尾、末尾多了话、把 Python 的 None/True 混进来。直接 json.loads 会炸。

思路: 别指望模型写对。建一条渐进降级的修复流水线:先做通用修复,再标准 json,再 json5,还不行就按字段类型逐个转义。能救则救,救不了给个带 raw 的错误对象让上层知道。

怎么读这张降级图(命中即停):

   模型手写的 arguments 字符串
            │
            ▼
  robust_json_loads(parsing_utils.py:188)
   ①preprocess:补未收尾的引号 + 智能截断多余内容
            │
            ▼
   ②json.loads  ──成功──► 返回 dict
            │失败
            ▼
   ③json5.loads ──成功──► 返回 dict(容忍单引号/尾逗号/注释)
            │失败(XML 分支才继续)
            ▼
   ④parse_escaped_json_string:5 级转义策略逐字段修
       直解 → 行首key → 反向前瞻key → legacy → 保守兜底
            │仍失败
            ▼
   ⑤ {"error":"Failed to parse arguments","raw": 原文}

通用预处理(preprocess_json_string,parsing_utils.py:162)做两件事:

  • _fix_unterminated_string_values(:68)——检测 "key": "值… 少了收尾引号的情况,在顶层收尾括号前补一个 ",再顺手补齐缺的 }/]。
  • _smart_json_truncate(:18)——从第一个 {/[ 开始按括号配平,在第一个完整结构结束处截断,把模型在 JSON 后面多说的话丢掉(且不数字符串内部的括号)。

分字段的智能修复(escape_string_content,parsing_utils.py:223)——按 key 名走不同策略,这个设计很贴合 agent 场景:

key 名修复函数干什么
code_blockfix_python_syntax(:290)把混进来的 null/true/false 改回 Python 的 None/True/False
commandfix_shell_syntax(:309)把 True/False/None 改成 shell 习惯的 true/false/""
其它fix_json_syntax(:327)把 Python 的 None/True/False 改回 JSON 的 null/true/false

换句话说:同一个 false,在代码字段里该是 Python 的 False,在 shell 命令里该是小写 true,在普通 JSON 里该是 false——修复流水线按字段语义分别纠正。这类"模型分不清语言关键字"的坑,是纯 JSON 解析器接不住的。

兜底诚实: 五级策略全挂了,parse_escaped_json_string 最后返回 _conservative_escape_fallback 只修最明显的换行问题(parsing_utils.py:559);再不行,arguments 直接变成 {"error": ..., "raw": arguments_str}(parsing_utils.py:767),把原文留给上层——不假装解析成功。

3.5 一个具体 provider:ClaudeAnthropicClient 的缓存控制​

它要解决的小问题: 多轮对话里,system prompt 和历史消息每轮都重复发送,很费 token。Anthropic 支持"prompt caching"——把稳定不变的前缀标记为可缓存,命中缓存的部分算钱便宜得多。

思路: 每次请求,把 system prompt 和最后一条 user 消息打上 cache_control: ephemeral 标记。稳定前缀(system + 早期历史)会被缓存复用,只有新增内容按全价计费。

真实实现: 发请求时 system 直接带缓存标记(claude_anthropic_client.py:77-83);历史侧由 _apply_cache_control(:215)只给"最后一条 user 消息"的文本块加标记:

# claude_anthropic_client.py:219 —— 摘录
for turn in reversed(messages):          # 从后往前找
    if turn["role"] == "user" and user_turns_processed < 1:
        ... text_item["cache_control"] = {"type": "ephemeral"}  # 只打最后一条 user
        user_turns_processed += 1

为什么打在"最后一条 user"? Anthropic 的缓存是"到这个断点为止的前缀都可缓存"。把断点放在最后一条 user 消息上,意味着除了这一轮模型即将生成的新内容,前面全都能走缓存——缓存收益最大化。

配套的 usage 记账: 命中缓存的 token 要单独统计,ClaudeAnthropicClient._extract_usage_from_response(:188)把 cache_creation + cache_read + input 合成总 input,并单列 cached_tokens = cache_read_input_tokens。基类的 get_usage_log(provider_client_base.py:368)再据此打出"Cached / Uncached"分项——方便算省了多少钱。

一个可靠性细节: _create_message 上挂了 @retry(wait=wait_fixed(10), stop=stop_after_attempt(5))(:47),API 抖动时固定间隔重试 5 次;asyncio.CancelledError 单独 re-raise 不吞。

3.6 token 治理:裁掉旧工具结果​

它要解决的小问题: agent 跑久了,历史里堆满巨大的工具返回(网页全文、命令输出),很快撑爆上下文。

思路: 保留对话结构(消息还在、轮次不乱),但把中间那些旧的工具结果内容替换成一句占位符,只留最早一条和最近 k 条的完整内容。

真实实现: _remove_tool_result_from_messages(provider_client_base.py:115)按 keep_tool_result 参数裁剪:

# provider_client_base.py:120 —— 摘录
user_indices = [i for i, msg in enumerate(messages_copy)
                if msg.get("role") in ("user", "tool")]
first_user_idx = user_indices[0]                 # 永远保留第一条(通常是原始任务)
num_to_keep = 0 if keep_tool_result == 0 else min(keep_tool_result, len(user_indices)-1)
last_indices_to_keep = user_indices[-num_to_keep:] if num_to_keep > 0 else []
indices_to_keep = [first_user_idx] + last_indices_to_keep
# 其余 user/tool 消息:
msg["content"] = "Tool result is omitted to save tokens."

关键设计点:

  • keep_tool_result == -1 = 不裁(:171),完整历史;正数才启用裁剪。
  • 第一条永远留 —— 它通常是原始任务描述,裁了 agent 就忘了自己在干嘛。
  • 裁的是 content 不是消息本身 —— 消息骨架、轮次都在,只是旧工具结果被换成占位符,模型仍知道"这里调过一次工具",只是看不到具体返回了。

裁剪在 Anthropic 的 _create_message 里先于缓存标记执行(claude_anthropic_client.py:63-67):先裁历史,再打缓存标记——顺序合理,免得给马上要被裁掉的内容打缓存。

注意:基类里还有一个更简单的 _filter_message_history(:254,create_message 入口调的那个)只是"取最后 k 条"的粗过滤;真正做"留头 + 留尾 + 中间打码"的精细裁剪是 _remove_tool_result_from_messages,由各 provider 的 _create_message 自己调。两者别混。

4. 巧妙之处(可借鉴的技术)​

  • 不信任 native tool_calls,改用文本协议 + 正则 —— 用"教模型写 <use_mcp_tool> XML 再正则抠"换来跨 provider 的统一与对弱模型的兼容(parsing_utils.py:584,prompt 见 main_agent_prompt_gaia.py:28)。妙在 prompt 里直说"我用正则解析、不要求合法 XML",主动降低对模型格式精度的依赖。
  • 按字段语义修 JSON —— code_block 用 Python 关键字、command 用 shell 关键字、其它用 JSON 关键字,同一个 false 按上下文修成三种写法(parsing_utils.py:277-285)。这是把"领域知识"塞进解析器的典型。
  • 漏标签自动补 + 递归重解析 —— 对最常见的"漏 </arguments>"补齐后递归再跑一次(parsing_utils.py:778-800),而不是直接报废整个调用。
  • 目录即插件注册表 —— providers/__init__.py 用 pkgutil.iter_modules 自动收类(:15),新增 provider 零注册成本;obj.__module__ == module.__name__ 精准排除 import 进来的类。
  • 缓存断点放最后一条 user 消息 —— 让"新生成内容之前的一切"都可缓存,收益最大化(claude_anthropic_client.py:219)。
  • 裁剪保头留尾、只打码内容 —— 省 token 但不破坏对话结构与任务记忆(provider_client_base.py:115)。

5. 边界与局限(诚实)​

  • 文本解析天然比 native 脆 —— 模型若把 <use_mcp_tool> 写进代码示例或引文里,正则可能误抓;好在 prompt 要求工具调用必须放在回复末尾、顶层来降低歧义(main_agent_prompt_gaia.py 的 Important Notes)。
  • 修复是启发式的,可能"修错" —— _fix_unterminated_string_values/智能截断都是猜测性修补,极端畸形输入下可能把本来的意思改坏;代码里多处用 try/except 兜底吞异常(如 parsing_utils.py:158),失败时退回原文。
  • 缓存/usage 依赖各家字段名 —— 一旦上游改字段(如 Anthropic 的 cache_read_input_tokens),_extract_usage_from_response 的统计就可能偏,需要跟着改子类。
  • XML 分支不产出 tool_use id —— str 分支里 id 恒为 None(parsing_utils.py:774),因为文本里本就没有 native 调用 id;要靠上层用顺序/内容对齐工具结果。
  • 裁剪逻辑对非 user/tool 角色不处理 —— 只按 role in ("user","tool") 裁(:123),assistant 里夹带的大段文本不在此机制覆盖内。

6. 代码地图(导航索引)​

主题文件路径符号名
provider 工厂(字符串→类)src/llm/client.py:13LLMClient
自动收集所有 provider 类src/llm/providers/__init__.py:15(模块级 pkgutil.iter_modules 循环)
抽象基类·统一门面src/llm/provider_client_base.py:27LLMProviderClientBase
统一入口:裁史+调子类+累计 usagesrc/llm/provider_client_base.py:177create_message
token 裁剪:留头留尾打码中间src/llm/provider_client_base.py:115_remove_tool_result_from_messages
usage 日志(Cached/Uncached 分项)src/llm/provider_client_base.py:368get_usage_log
工具名注册为 server-toolsrc/llm/provider_client_base.py:221convert_tool_definition_to_tool_call
Anthropic 发请求 + 重试src/llm/providers/claude_anthropic_client.py:47_create_message
Anthropic 缓存控制src/llm/providers/claude_anthropic_client.py:215_apply_cache_control
Anthropic usage 解析src/llm/providers/claude_anthropic_client.py:188_extract_usage_from_response
三形态工具调用解析器src/utils/parsing_utils.py:584parse_llm_response_for_tool_calls
健壮 JSON 加载(json→json5)src/utils/parsing_utils.py:188robust_json_loads
JSON 预处理(补引号+截断)src/utils/parsing_utils.py:162preprocess_json_string
补未收尾字符串值src/utils/parsing_utils.py:68_fix_unterminated_string_values
智能截断多余内容src/utils/parsing_utils.py:18_smart_json_truncate
按字段语义转义/修关键字src/utils/parsing_utils.py:223escape_string_content
5 级转义修复src/utils/parsing_utils.py:340parse_escaped_json_string
教模型输出 XML 的 promptconfig/agent_prompts/main_agent_prompt_gaia.py:28(system prompt 文本)

相关章节: 编排主循环 · 分层 sub-agent · 工具层:MCP server 池 · 配置矩阵与容错哲学 · 总览与阅读地图