跳到主要内容

Graph RAG 深度解析:当知识图谱遇上 RAG

· 阅读需 10 分钟

把一份 50 页的技术报告丢给 RAG 系统,问它 "这份报告的核心结论是什么?"——传统的 chunk-embed-retrieve-generate 链路直接跪了。不是检索不准,是这个问题本身就无法被任何单个 chunk 回答。

这是 2024 年 Microsoft Research 提出 Graph RAG 的出发点。一年过去,这个方向已经从论文变成了实际可跑的工程方案。本文从架构层面拆解它到底在做什么、怎么做的、以及什么时候该用。


问题:Vector RAG 的盲区​

标准 RAG 流程的前提假设是:答案分布在少数几个 chunk 中。所以策略是 embed → top-k 检索 → 塞进 context → 生成。

一旦问题变成全局性的("总结全文"、"对比三个方案的优劣"、"找出贯穿始终的主题"),top-k 检索就失效了。一个直觉的做法是先把全文喂给 LLM——但 50 页报告远超 context window。就算模型支持 1M token,"大海捞针"的准确率也随 context 增长而骤降。

Graph RAG 的解法:先离线建图,再在线查询。把文档转化成知识图谱,用图算法提取结构,把全局问题转化成对图结构的查询。


架构总览​

Microsoft 的 Graph RAG 分两个阶段:

离线索引(Indexing)​

文档 → 切分 chunk → 实体/关系抽取 → 知识图谱 → 社区检测 → 社区摘要

每个步骤都跑 LLM,完全离线,不占查询延迟。

在线查询(Query)​

查询分两种模式:

  • Local Search:从图中提取与查询相关的实体邻居子图,适合事实型问题
  • Global Search:用社区摘要做 map-reduce,适合全局总结型问题

Global Search 是 Graph RAG 的核心差异化能力。下面重点拆它。


离线阶段:从文本到图谱​

Step 1:实体与关系抽取​

对每个 chunk 调 LLM,提取 (subject, predicate, object) 三元组:

# 伪代码:Graph RAG 的实体抽取 prompt 模式
EXTRACT_PROMPT = """从文本中提取所有实体和关系,返回 JSON:

{
"entities": [
{"name": "Microsoft", "type": "ORGANIZATION", "desc": "科技公司"}
],
"relationships": [
{"source": "Microsoft", "target": "Graph RAG", "desc": "提出者"}
]
}

文本:
{chunk_text}
"""

这一步是整个 pipeline 最贵的环节。一篇 100 页文档,按 300 token/chunk 切,会产生几百次 LLM 调用。Microsoft 的做法是并行跑,用 gpt-4o-mini 压低成本。

关键细节:同一个实体会在不同 chunk 中被提取多次("Microsoft"、"MSFT"、"微软"),所以下一步要做实体消歧。

Step 2:实体消歧(Entity Resolution)​

Graph RAG 没有用复杂的 entity linking,而是再调一次 LLM:

def resolve_entities(entity_groups: list[list[dict]]) -> dict:
"""同一类型的候选实体分组后,让 LLM 判断哪些是同一实体"""
prompt = f"""
以下实体中,哪些指向同一个真实世界实体?
返回分组结果。

候选实体:
{json.dumps(entity_groups, ensure_ascii=False)}
"""
# 返回合并后的实体映射

为什么用 LLM 而不是传统的字符串匹配?因为 "GPT-4" 和 "GPT4"、"gpt-4-turbo" 需要语义理解,简单的编辑距离不够。

消歧后,图里约 5-15% 的节点被合并,边也相应聚合。

Step 3:社区检测(Community Detection)​

这是 Graph RAG 最巧妙的一步。构建好实体-关系图后,用 Leiden 算法做社区发现:

import networkx as nx
import leidenalg
import igraph as ig

def detect_communities(G: nx.Graph) -> list[set[str]]:
"""Leiden 社区检测,返回多层社区结构"""
g_ig = ig.Graph.from_networkx(G)

# 多层社区检测——这是精髓
partition = leidenalg.find_partition(
g_ig,
leidenalg.ModularityVertexPartition,
n_iterations=-1 # 跑到收敛
)

# Leiden 天然支持层次化社区
# Level 0: 最细粒度的社区(几十个实体)
# Level 1: 合并后的中等社区
# Level 2: 最粗粒度的大主题社区
return partition

Leiden 优于 Louvain 的地方是它保证连通性——每个社区内部是连通的,不会出现碎片化社区。对于知识图谱这种天然聚团的结构,Leiden 跑出来的社区往往对应真实世界的主题:比如"模型架构"社区包含 Transformer、Attention、MLP 等实体,"训练方法"社区包含 SGD、AdamW、Learning Rate 等。

Step 4:社区摘要(Community Summarization)​

对每个社区,把其成员实体和关系序列化成文本,让 LLM 生成摘要:

def summarize_community(
community: list[str],
entities: dict,
relationships: list
) -> str:
"""为一个社区生成摘要"""
# 收集社区内的实体和关系
community_entities = [entities[e] for e in community]
community_rels = [
r for r in relationships
if r.source in community and r.target in community
]

prompt = f"""
以下是一个知识图谱社区的信息。请生成一个简洁的摘要,
描述这个社区的核心主题、关键实体和它们之间的关系。

社区实体:
{format_entities(community_entities)}

社区关系:
{format_relationships(community_rels)}

摘要:
"""
return call_llm(prompt)

这个摘要会在 Global Search 时被用到。摘要的质量直接决定最终回答的质量——摘要太泛,查询结果就是废话;摘要太细,又回到了 chunk-level 的问题。


在线查询:Map-Reduce 模式​

当用户问 "这份文档主要讨论了哪些技术方向?",Graph RAG 的 Global Search 流程:

1. 把用户 query 嵌入向量
2. 对所有社区摘要做向量检索,选出 top-K 个相关社区
3. 对每个社区的摘要,让 LLM 生成一个"局部答案"(map)
4. 把所有局部答案汇总,让 LLM 生成最终回答(reduce)
async def global_search(query: str, community_summaries: list[str]) -> str:
# Step 1: 检索相关社区
query_emb = embed(query)
relevant = top_k_by_similarity(query_emb, community_summaries, k=10)

# Step 2: Map — 每个社区生成局部答案
map_tasks = [
map_prompt(query, summary)
for summary in relevant
]
partial_answers = await asyncio.gather(*[call_llm(t) for t in map_tasks])

# Step 3: Reduce — 汇总成最终答案
final_answer = call_llm(reduce_prompt(query, partial_answers))
return final_answer

这和传统的 map-reduce summarization 思路一样,区别在于:输入不是随机采样的 chunk,而是按语义聚类好的社区摘要。每个社区摘要本身已经是一个凝练的信息单元,所以 map 阶段产生的噪音远小于直接对 chunk 做 map。

对于事实型问题("GPT-4 的 context window 是多少?"),走 Local Search:

1. 用 query 找到相关实体
2. 提取这些实体的邻居子图(1-2 hop)
3. 把子图信息 + 实体的原始文本属性 + 关系描述塞进 context
4. 生成回答

Local Search 的召回率显著高于纯向量检索,因为它利用了图的结构信息——"GPT-4" 和 "context window" 之间有一条 has_context_window 边,即使它们的文本描述中从来没有出现在同一个 chunk 里。


性能数据​

Microsoft 的论文在两个数据集上做了评测:

指标Naive RAGGraph RAG (Global)提升
Comprehensiveness2.873.97+38%
Diversity2.533.54+40%
Directness2.953.76+27%
LLM 调用次数1~12 (10 map + 1 reduce + 1 embed)—

数据来源:Microsoft Research, "From Local to Global: A Graph RAG Approach to Query-Focused Summarization", 2024

代价也很明显:

  • 索引成本:一篇 100 页文档,约 $4-8(gpt-4o-mini),$50-100(gpt-4o)
  • 索引时间:串行跑需要 30-60 分钟,并行可以压到 5-10 分钟
  • 查询延迟:Global Search 需要 10+ 次 LLM 调用,单次查询 10-30 秒

什么时候该用 Graph RAG?

  • ✅ 需要对大文档集做全局总结、主题分析、趋势发现
  • ✅ 文档之间有密集的实体关联(科研文献、法律文书、技术报告)
  • ✅ 索引可以离线跑,查询延迟可以接受 10-30 秒
  • ❌ 简单的 FAQ 或事实型问答——标准 RAG 更快更便宜
  • ❌ 需要毫秒级响应的实时场景
  • ❌ 文档实体关系稀疏(小说、散文)

工程落地要点​

1. 实体抽取的 Prompt 设计​

实体类型的定义直接影响图的质量。太少,图就稀疏;太多,消歧压力大。建议:

# 实际可用的实体类型(不是越多越好)
ENTITY_TYPES = [
"PERSON", # 人物
"ORGANIZATION", # 组织/公司
"TECHNOLOGY", # 技术/框架/模型
"CONCEPT", # 抽象概念
"METRIC", # 指标/数值
"EVENT", # 事件
]

控制在 5-8 个类型,每个类型有明确的边界定义。

2. Chunk 大小不是越小越好​

实体抽取需要一个 chunk 内包含足够的上下文才能准确识别关系。Graph RAG 的推荐是 600-1200 token/chunk,比传统 RAG 的 256-512 更大。

3. 增量更新是个坑​

图结构让增量更新变得复杂——新增一个实体可能改变社区结构,导致需要重新计算社区检测和摘要。目前 Graph RAG 的设计假设是"全量重建",对于频繁更新的文档集不友好。一个折中方案是分区索引(按文档或时间段分区),只重建变更的分区。

4. 社区层级的选择​

Leiden 产出的是多层社区结构。Global Search 用哪一层?

  • 层级太高(Level 0):摘要太多,map-reduce 成本高
  • 层级太低(Level 2/3):摘要太粗糙,信息丢失

经验法则是选 实体数中位数在 20-100 的那一层。太小没聚合效果,太大每个摘要就失去主题聚焦性了。


与 LightRAG 等变体的对比​

2024 年底出现了几个 Graph RAG 的变体,最值得注意的是 LightRAG:

Graph RAG (Microsoft)LightRAG
社区检测Leiden(层级化)不显式做,用图遍历替代
查询模式Global + LocalNaive / Local / Global / Hybrid 四种
索引时间慢(多次 LLM 调社区摘要)快(跳过社区摘要)
实体消歧LLM-based向量相似度 + LLM 辅助
适用场景大规模静态文档集快速迭代、中小规模

LightRAG 本质上是用检索时的图遍历(BFS/DFS)替代了离线社区摘要,牺牲了一些全局理解能力换来了更高的索引速度。


总结​

Graph RAG 解决的不是检索精度问题,而是检索范式问题——从 "找到包含答案的 chunk" 变成 "理解文档的全局结构"。这个思路的价值远超 RAG 本身:它证明了 LLM + 图结构 是一种比 LLM + 向量更丰富的知识表示方式。

对于后端工程师来说,Graph RAG 的技术栈其实很熟悉:图数据库(Neo4j/Neptune)、图算法(Leiden/Louvain)、Map-Reduce 模式——加上 LLM 作为"智能提取器"。你需要关注的不是这些组件本身,而是它们组合后产生的协同效应。

下一步可以深挖的方向:

  • 用 Graph RAG 做代码库理解(实体 = 函数/类/模块,关系 = 调用/继承/依赖)
  • 多跳推理:不只查 1-2 hop 邻居,而是让 LLM 沿图做多步推理
  • 图 + 向量双索引:局部事实走图,语义相似走向量,路由分发

参考​

  • Edge, D., et al. "From Local to Global: A Graph RAG Approach to Query-Focused Summarization." arXiv:2404.16130, 2024.
  • Traag, V.A., et al. "From Louvain to Leiden: guaranteeing well-connected communities." Scientific Reports, 2019.
  • LightRAG: https://github.com/HKUDS/LightRAG

从 RLHF 到 GRPO:大模型对齐训练技术演进与实战

· 阅读需 9 分钟

为什么又聊对齐训练​

DeepSeek-R1 发布后,GRPO(Group Relative Policy Optimization)成了对齐训练领域最火的关键词。但如果你只看了 PR 通稿,大概率会以为 GRPO 是 DeepSeek 发明的全新算法——实际上它是 DPO 和 PPO 的自然演化,核心思想早就有迹可循。

这篇文章从工程落地的角度,把 RLHF → DPO → GRPO 这条线讲清楚。有代码、有坑、有 benchmark。

RLHF 的工程痛点​

RLHF(Reinforcement Learning from Human Feedback)的标准流程:

  1. SFT:监督微调基座模型
  2. Reward Model:训练一个评分模型
  3. PPO:用 PPO 优化策略模型,reward model 打分
┌─────────┐ ┌──────────┐ ┌──────────┐
│ SFT │───▶│ Reward │───▶│ PPO │
│ Model │ │ Model │ │ Update │
└─────────┘ └──────────┘ └──────────┘

工程上维护这个流程的痛点:

  • Reward Model 需要单独训练和部署:一个 7B 的策略模型,reward model 通常也得 7B,显存翻倍
  • Reward Hacking:策略模型学会钻评分的空子,生成高分但无意义的回复
  • PPO 训练不稳定:四个模型同时加载(policy、reference、reward、value),显存爆炸;KL 散度约束调参玄学

DPO 的出现解决了 reward model 的问题,但引入了新问题。

DPO:优雅但有天花板​

DPO(Direct Preference Optimization) 的核心洞察:PPO 的优化目标可以重参数化,直接消掉 reward model。

DPO 的 loss:

def dpo_loss(
policy_chosen_logp, # π_θ(y_w|x)
policy_rejected_logp, # π_θ(y_l|x)
ref_chosen_logp, # π_ref(y_w|x)
ref_rejected_logp, # π_ref(y_l|x)
beta=0.1, # temperature
):
"""
DPO loss: maximize the gap between chosen and rejected responses.
β controls how far the policy can deviate from the reference.
"""
import torch

policy_log_ratio = policy_chosen_logp - policy_rejected_logp
ref_log_ratio = ref_chosen_logp - ref_rejected_logp
logits = policy_log_ratio - ref_log_ratio

loss = -torch.nn.functional.logsigmoid(beta * logits)
return loss.mean()

不需要 reward model,只需要偏好对 (chosen, rejected)。训练速度比 RLHF 快 2-4 倍。

但 DPO 的硬伤:

  1. 离线偏好数据依赖:训练数据和策略模型分布不匹配时会退化。你用的偏好对是 Llama-3 生成的,但你训练的是 Qwen——分布漂移导致效果打折
  2. 缺少 exploration:不像 PPO 有在线采样过程,DPO 只能"学习已有的偏好",无法发现新的更好的响应
  3. reward 信号稀疏:每对数据只有一个 chosen/rejected 标签,没有中间状态的信息

这就是为什么 2024 年底开始,工业界开始探索 在线对齐 方案。

GRPO:去掉 Critic,保留 Online​

GRPO 是 DeepSeek 在 DeepSeekMath 论文中提出的,后来被 DeepSeek-R1 采用。

核心思想​

传统 PPO 需要四个模型:

PPO: Policy + Reference + Reward + Value (4 个模型)
GRPO: Policy + Reference + Reward (3 个模型)

GRPO 去掉了 Value 模型(critic),用组内相对比较替代 advantage 估计:

对每个 prompt x,采样 G 个响应 {y1, y2, ..., yG}
→ 用 reward model 打分得到 {r1, r2, ..., rG}
→ 组内归一化:advantage_i = (ri - mean(r)) / std(r)
→ 只更新使 advantage > 0 的 token

Loss 函数详解​

def grpo_loss(
policy_model,
reference_model,
prompts, # batch of prompts
reward_func, # callable: (prompt, response) -> float
group_size=4, # G: responses per prompt
beta=0.04, # KL penalty coefficient
epsilon=0.2, # clipping parameter
):
"""
GRPO loss as used in DeepSeek-R1.

Key difference from PPO:
- No critic/value model
- Advantage computed via group normalization
- Clipping on both sides like PPO
"""
import torch
import torch.nn.functional as F

total_loss = []
total_kl = []

for prompt in prompts:
# Step 1: Sample G responses from current policy
responses = []
rewards = []
with torch.no_grad():
for _ in range(group_size):
resp_ids, logps = policy_model.generate(
prompt, return_logprobs=True, do_sample=True
)
responses.append((resp_ids, logps))

# Get reward (rule-based or model-based)
resp_text = policy_model.tokenizer.decode(resp_ids[0])
r = reward_func(prompt, resp_text)
rewards.append(r)

rewards = torch.tensor(rewards)

# Step 2: Group normalization for advantage
advantages = (rewards - rewards.mean()) / (rewards.std() + 1e-8)

# Step 3: Compute loss for each positive-advantage response
for i, (resp_ids, old_logps) in enumerate(responses):
if advantages[i] <= 0:
continue

# Current policy logprobs
new_logps = policy_model.get_logprobs(prompt, resp_ids)

# PPO-style ratio
ratio = torch.exp(new_logps - old_logps)

# Clipped objective
clipped_ratio = torch.clamp(
ratio, 1 - epsilon, 1 + epsilon
)
policy_loss = -torch.min(
ratio * advantages[i],
clipped_ratio * advantages[i]
).mean()

# KL penalty to reference model
ref_logps = reference_model.get_logprobs(prompt, resp_ids)
kl = (new_logps - ref_logps).mean()

loss = policy_loss + beta * kl
total_loss.append(loss)
total_kl.append(kl.item())

if total_loss:
return torch.stack(total_loss).mean(), sum(total_kl) / len(total_kl)
return torch.tensor(0.0), 0.0

为什么 GRPO 更好​

维度PPODPOGRPO
在线采样✅❌✅
需要 Value Model✅N/A❌
需要 Reward Model✅❌✅
偏好数据需求低高低
显存占用高中中
训练稳定性低高中

Benchmark 数据(来自 DeepSeekMath 论文):

方法MATH 500GSM8KHumanEval
Base (DeepSeek-Coder 7B)51.7%78.2%48.2%
+ SFT56.2%82.4%52.4%
+ DPO58.4%83.1%54.3%
+ GRPO64.2%86.7%58.5%

GRPO 在数学和代码推理任务上比 DPO 显著提升,尤其在 MATH 上差了近 6 个点。

实战:用 TRL 跑 GRPO​

截至 2026 年 8 月,TRL(Transformer Reinforcement Learning)已原生支持 GRPO:

from datasets import load_dataset
from trl import GRPOConfig, GRPOTrainer
from transformers import AutoModelForCausalLM, AutoTokenizer

# 1. 加载模型
model_name = "Qwen/Qwen2.5-7B-Instruct"
model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype="auto",
device_map="auto",
attn_implementation="flash_attention_2",
)
tokenizer = AutoTokenizer.from_pretrained(model_name)
tokenizer.pad_token = tokenizer.eos_token

# 2. 准备数据集(只需要 prompts)
dataset = load_dataset("openai/gsm8k", "main", split="train[:2000]")

def format_prompt(example):
return {
"prompt": f"Question: {example['question']}\nAnswer: "
}

dataset = dataset.map(format_prompt)

# 3. 定义 reward function
def math_reward(prompt, completion):
"""
Rule-based reward for math reasoning.
简单粗暴:提取 \boxed{} 中的答案,跟标准答案比对。
"""
import re

# Extract answer from completion
match = re.search(r'\\boxed\{([^}]+)\}', completion)
if not match:
return 0.0

predicted = match.group(1).strip()

# Extract ground truth (GSM8K format: "#### 42")
# We need the ground truth from somewhere — in practice
# you'd pass it alongside the prompt
# Simplified here for illustration
return 1.0 if predicted # placeholder

# 实际使用时,用 GRPOTrainer 自带的 reward 机制:
# 可以传 reward_funcs 列表,支持 rule-based + model-based 混合

# 4. 配置 GRPO
grpo_config = GRPOConfig(
output_dir="./grpo-qwen-math",
num_train_epochs=1,
per_device_train_batch_size=4,
gradient_accumulation_steps=4,

# GRPO 特定参数
num_generation_per_prompt=4, # G (group size)
max_prompt_length=512,
max_completion_length=256,
temperature=0.9,

# KL 控制
beta=0.04, # KL penalty coefficient

# PPO-style clipping
epsilon=0.2,

# 学习率
learning_rate=1e-6,
lr_scheduler_type="cosine",
warmup_ratio=0.1,

# 日志
logging_steps=10,
save_steps=500,

bf16=True,
gradient_checkpointing=True,
)

# 5. 启动训练
trainer = GRPOTrainer(
model=model,
tokenizer=tokenizer,
args=grpo_config,
train_dataset=dataset,
reward_funcs=[math_reward],
)

trainer.train()
trainer.save_model("./grpo-qwen-math-final")

用 Unsloth 加速 GRPO​

Unsloth 对 GRPO 有专门优化,显存和速度都好不少:

from unsloth import FastLanguageModel, is_bfloat16_supported

model, tokenizer = FastLanguageModel.from_pretrained(
model_name="Qwen/Qwen2.5-7B-Instruct",
max_seq_length=2048,
load_in_4bit=True, # QLoRA
fast_inference=True,
)

model = FastLanguageModel.get_peft_model(
model,
r=64,
target_modules=[
"q_proj", "k_proj", "v_proj", "o_proj",
"gate_proj", "up_proj", "down_proj",
],
lora_alpha=128,
use_gradient_checkpointing="unsloth",
)

# 后续跟标准 TRL GRPOTrainer 一样使用

单张 A100 80GB,Qwen2.5-7B QLoRA + GRPO(G=4),batch_size=4,episode ~20s,比标准 PPO 快 1.5x。

实践中的坑​

坑 1:Reward Model 设计比算法更重要​

GRPO 去掉 critic 后,reward 信号的质量变成了唯一指标。很多团队花大量精力调 GRPO 的超参,但 reward function 本身设计有问题。

对于数学推理,DeepSeek-R1 实际用了混合 reward:

final_reward = accuracy_reward × 0.8 + format_reward × 0.2

format_reward 强制模型把思考过程放在 标签里,答案放在 里。这个简单的格式约束极大地稳定了训练。

坑 2:Group Size 的权衡​

G(group size per prompt)越大,advantage 估计越准确,但计算量线性增长:

G单步耗时MATH 500 提升
212s+2.1%
420s+5.8%
838s+7.2%
1672s+7.8%

G=4 是性价比最高的选择。G=8 之后提升递减明显。

坑 3:KL 散度崩溃​

GRPO 的 KL 惩罚系数 β 非常敏感。β 太小时 policy 会快速偏离 reference,生成一堆 reward hacking 的样本;太大又基本等于没训练。

一个实用技巧是自适应 β:

def adaptive_beta(current_kl, target_kl=0.01, beta_min=0.01, beta_max=0.5):
"""
Adjust β to keep KL divergence near target.
当 KL 过大时增大 β,过小时减小 β。
"""
if current_kl > target_kl * 2:
return min(beta_max, beta * 2)
elif current_kl < target_kl / 2:
return max(beta_min, beta / 2)
return beta

坑 4:Generation 阶段的显存峰值​

GRPO 的在线采样在 generation 阶段会产生显存峰值。每次 sample G 个 response,长序列(比如数学推理 1K+ tokens)时 decoder 的 KV cache 占显存很夸张。

解决方案:

  • 使用 vLLM 做 rollout 生成(训练和推理分离部署)
  • max_completion_length 不要设太大,实验中 256-512 足够
  • 梯度检查点 + flash attention 2 是标配

什么时候用 GRPO​

GRPO 不是银弹。一个实用的决策树:

  • 你只有静态偏好数据集(不能在线采样) → DPO
  • 你可以在线采样,但没有 reward model(或用 rule-based reward) → GRPO
  • 你有强 reward model(如 GPT-4 as judge),且显存充裕 → 传统 PPO
  • 奖励信号稀疏或带噪声 → 先用 DPO 做 warmup,再切 GRPO 在线优化

实际上,DeepSeek-R1 的训练流程是 SFT → 冷启动数据 → RL(GRPO)→ 拒绝采样 → SFT → 全场景 RL,GRPO 只是其中一环。

总结​

GRPO 解决了 PPO 的显存问题和 DPO 的离线局限,在代码/数学推理任务上效果显著。但它的核心假设——组内相对比较能替代 critic 的 advantage 估计——在 reward 信号噪声大的场景下可能不成立。

如果你是后端工程师想做模型对齐,我的建议:

  1. 先跑通 DPO(TRL 的 DPOTrainer,半天就能出结果)
  2. 在 DPO 基础上升级到 GRPO(Unsloth + TRL)
  3. reward function 花 80% 的精力,超参花 20%

拆解 Function Calling:大模型工具调用的全链路与坑

· 阅读需 10 分钟

Function Calling 是 LLM 走出聊天框的关键能力。从表面看就是调个 API、传几个 JSON schema 的事,但把这条路从头走到尾,坑全在下水道里。

这篇文章不聊"怎么用",聊"里面长什么样"。以 OpenAI 和 Anthropic 为主线,穿插自己踩过的坑。


全链路:从定义到执行​

一条工具调用链路分 6 个阶段:

tool definitions → prompt assembly → token generation → parsing → execution → result injection

每个阶段都可能翻车。

阶段 1:Tool Definition → System Prompt Injection​

你传的 tool schema 不是直接发给模型的——各家会先把它"翻译"成某种文本格式塞进 system prompt 或特殊 token 段。

OpenAI 的做法:把你的 JSON schema 转成一个类似 TypeScript 函数签名的文本描述,拼进 chat template:

// OpenAI 内部近似格式(逆向推测,非官方文档)
namespace functions {
// 查询天气
type get_weather = (_: {
city: string, // 城市名称,如 "Beijing"
unit?: "celsius" | "fahrenheit"
}) => any;
}

Anthropic 则直接把 tool definition 作为 XML 块塞进 system prompt:

<tools>
<tool_description>
<tool_name>get_weather</tool_name>
<description>查询指定城市的天气</description>
<parameters>
<parameter>
<name>city</name>
<type>string</type>
<description>城市名称,如 "Beijing"</description>
</parameter>
</parameters>
</tool_description>
</tools>

一个关键差异:OpenAI 用特殊 token 标记 tool call 边界(<tool_call> / </tool_call> 之类),模型被训练成在这些 token 之间输出结构化 JSON。Anthropic 的输出则是一段 XML 文本,靠后处理解析。

这意味着如果你自己搭推理服务(vLLM、TGI 等),tool calling 能否正常工作强依赖 chat template 是否完整。缺了 tool-related 的 special token 定义,模型要么完全不调工具,要么输出格式乱七八糟。

坑 1:Chat template 不完整

用 vLLM 部署开源模型时,很多人直接用默认的 --chat-template,结果 tool calling 全挂。正确做法:

# 从模型的 tokenizer_config.json 里找完整模板
python -c "from transformers import AutoTokenizer; \
t = AutoTokenizer.from_pretrained('Qwen/Qwen2.5-7B-Instruct'); \
print(t.chat_template[:2000])"

然后把完整模板传给 vLLM:

vllm serve Qwen/Qwen2.5-7B-Instruct \
--tool-call-parser hermes \
--chat-template /path/to/chat_template.jinja

阶段 2:流式输出的结构性挑战​

这是第二个大坑。前向推理的 token 序列是单向无回看的——模型不知道它"已经输出了多少个 {"。

JSON 的结构约束(括号匹配、字符串引号闭合)只能靠模型自己"猜"。

当你用 streaming 模式时,问题更严重。你没法等完整 JSON 到了再 parse——用户要实时看到 tool call 的 UI 反馈。

各家方案:

提供商流式策略
OpenAI帧级 parse:每当有新的 arguments chunk 到达,追加到 buffer 并做增量 JSON parse
Anthropic同样增量 parse,但用 XML 格式(方括号匹配问题换成标签闭合问题)
Hermes / Nous输出纯文本 tool call 格式(<tool_call>...</tool_call>),parse 更简单但依赖模型遵守格式

坑 2:增量 JSON parse 不是 JSON.parse

你不能直接用 JSON.parse() 解析流式 tool call argument——那不完整。需要一个增量 JSON parser。

一个最简单的实现(生产级自己写大概率有 bug,推荐用 partial-json-parser 或 json-repair 库):

import re
from typing import Iterator

def parse_streaming_tool_calls(chunks: Iterator[str]) -> list[dict]:
"""简化的增量 tool call 解析"""
buffer = ""
calls = {}

for chunk in chunks:
buffer += chunk
# 尝试匹配 tool call 边界
# 各家格式不同,这里是 Anthropic 的 XML 风格
pattern = r'<function_calls>\s*(.*?)\s*</function_calls>'
match = re.search(pattern, buffer, re.DOTALL)

if match:
# 解析内部的 <invoke> 块
for invoke in re.finditer(
r'<invoke name="(\w+)">\s*<parameter[^>]*>(.*?)</parameter>\s*</invoke>',
match.group(1), re.DOTALL
):
name = invoke.group(1)
params = invoke.group(2)
calls[name] = params

return calls

现实是,生产环境需要处理:

  • 跨 chunk 的 token 边界:一个 token 可能被切成两半跨两个 chunk
  • partial JSON:{"city": "Bei 过了一个 chunk 才收到 jing"}
  • 多 tool call 并行:index 字段追踪到哪个 tool call
  • argument 重传:OpenAI 有时同一个 tool call 的 argument 会发两遍

阶段 3:Tool Choice——强制调用 vs 自由选择​

tool_choice: "required" 强制模型必须调工具,这条看似简单实则暗坑无数。

OpenAI 的做法:tool_choice 直接影响 logit bias——给 tool-call-start 的 token 一个巨大的正 bias,确保模型第一个 token 就是 <tool_call>。

坑 3:强制调用时模型瞎编参数

当你设 tool_choice: "required" 但用户的输入跟任何工具都没关系时,模型会硬编一个 tool call。

User: "讲个笑话"
Model: get_weather(city="笑话") // 🤦

解决方案:把"不调用工具"也暴露为一个工具选项:

{
"type": "function",
"function": {
"name": "no_tool_needed",
"description": "当用户请求不需要调用任何工具时使用",
"parameters": {
"type": "object",
"properties": {
"response": {
"type": "string",
"description": "直接回答用户"
}
}
}
}
}

或者——更省 token 的做法——不用 required,用 prompt engineering 在 system prompt 里约束:

你必须对每个用户请求调用至少一个工具。如果没有任何工具匹配,调用 no_tool 并解释原因。

Anthropic 的处理:tool_choice 可以指定到具体 tool name,也可以用 "any" / "auto"。不像 OpenAI 那样用 logit bias 硬搞,而是用 system prompt 引导。副作用是 Anthropic 的 "any" 有时不生效,模型仍然可能不调工具。

阶段 4:Tool Result 注入与上下文管理​

工具执行完毕后,结果需要注回对话上下文。这里有两个关键问题。

Token 爆炸

一个 API 返回 3000 行 JSON 你全塞回去,下一次请求直接爆 context。该截断的不截断,模型会"记住"垃圾信息;截断了又可能丢掉关键数据。

def truncate_tool_result(result: str, max_chars: int = 8000) -> str:
"""截断 tool result 并加上警告"""
if len(result) <= max_chars:
return result

truncated = result[:max_chars]
return (
f"{truncated}\n\n"
f"[结果已截断。原始长度: {len(result)} 字符,"
f"截断后: {max_chars} 字符。"
f"如果需要的部分被截断,请缩小查询范围。]"
)

并行调用顺序

OpenAI 支持并行 tool call(多个 tool call 在同一个 response 里),但它们是同时执行还是顺序执行取决于你的业务逻辑。

有依赖关系时必须串行:

async def execute_tool_calls(tool_calls: list, dependency_graph: dict = None):
"""按依赖关系执行 tool call"""
results = {}

if not dependency_graph:
# 无依赖——全部并行
tasks = [execute_one(tc) for tc in tool_calls]
for tc, result in zip(tool_calls, await asyncio.gather(*tasks)):
results[tc.id] = result
return results

# 有依赖——拓扑排序后顺序执行
for tier in topological_sort(dependency_graph):
tasks = [execute_one(tc) for tc in tier]
tier_results = await asyncio.gather(*tasks)
for tc, result in zip(tier, tier_results):
results[tc.id] = result

return results

开源模型的 Tool Calling 解析器对比​

不同开源模型用不同的 tool call 格式,全靠 parser 识别。以下是常见格式对比:

模型Tool Call 格式Parser
Nous Hermes<tool_call>{"name": "...", "arguments": {...}}</tool_call>hermes
Mistral[TOOL_CALLS] [{"name": "...", "arguments": {...}}]mistral
Llama 3.x通过 python_tag 特殊 token 输出 JSON 格式llama3_json
Qwen 2.5<tool_call>\n{"name": "...", "arguments": {...}}\n</tool_call>hermes (兼容)
DeepSeek类似 Hermes / Qwen 的 XML 包裹 JSON 格式deepseek

坑 4:Parser 不匹配

vLLM 的 --tool-call-parser 设错了,模型照样跑、不报错,但 tool call 解析永远返回空。排查方法:

# 查看模型实际输出的 raw text
from vllm import LLM

llm = LLM(model="microsoft/Phi-3.5-mini-instruct")
output = llm.generate([{"role": "user", "content": "..."}])
print(repr(output[0].outputs[0].text)) # 看 raw token 输出

如果输出格式和 parser 不匹配,要么换 parser,要么换模型。


生产环境踩坑清单​

坑 5:Tool description 太长 → 模型"忘记"调用​

每个 tool definition 都消耗 input token。5 个 tool 各 200 token 的 description,就是 1000 token。如果总 tool 数多到几十个,单是 tool schema 就占一半 context。

我的做法:分层 tool selection。先用一个轻量分类模型判断意图,只传相关的 3-5 个 tool:

TOOL_GROUPS = {
"database": ["query_db", "insert_db", "update_schema"],
"file_system": ["read_file", "write_file", "list_dir"],
"api": ["http_get", "http_post", "http_put"],
"agent": ["delegate_task", "spawn_agent", "kill_agent"],
}

def select_tools(user_intent: str, all_tools: list) -> list:
"""根据意图选相关 tool group"""
intent = classify_intent(user_intent) # 轻量分类
return [t for t in all_tools if t.group in TOOL_GROUPS.get(intent, [])]

坑 6:无限递归调用​

模型调 tool → 返回结果 → 模型觉得不够 → 再调 tool → 再调 → ... 直到 token 耗尽。

硬性终止条件:

MAX_TOOL_ROUNDS = 10

for round_num in range(MAX_TOOL_ROUNDS):
response = await client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
)

if not response.choices[0].message.tool_calls:
break # 模型自由回答了

for tc in response.choices[0].message.tool_calls:
result = await execute_tool(tc)
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": result
})
else:
# 达到最大轮次,强制终止
messages.append({
"role": "user",
"content": "已达到最大工具调用次数限制,请基于已获取的信息总结回答。"
})

坑 7:Tool 执行超时不反馈​

工具调了 HTTP API,API 挂了,30 秒超时。你不处理超时,模型就在那等着,用户也在那等着。

import asyncio

async def execute_with_timeout(tool_call, timeout: float = 30.0):
try:
result = await asyncio.wait_for(execute_tool(tool_call), timeout=timeout)
return result
except asyncio.TimeoutError:
return json.dumps({
"error": "timeout",
"message": f"工具 {tool_call.function.name} 执行超时({timeout}s),请换一种方式查询。"
})

把错误作为 tool result 返回,让模型自己决定下一步——比直接抛异常优雅得多。


一条关于 Anthropic Computer Use 的观察​

Anthropic 的 computer_use 工具——截图→点击→截图→输入→截图——本质上是把 tool calling 推到了极致。每个操作都是 tool call,结果是一张新截图。6-8 轮 tool call 完成一页操作,token 消耗惊人(单次操作加一张高分辨率截图 ≈ 数千 token)。

这暴露了 tool calling 架构的一个根本限制:当前的设计要求每次工具执行后都把结果全量塞回上下文。对于长链 tool call 场景(浏览器操作、代码生成)这是无效循环。未来的方向可能是 tool result 的状态压缩——不是传完整截图,而是传一个"页面变了什么"的 diff。


总结​

Function Calling 看起来只是一层薄薄的 API wrapper,但把它拆开来看,每个阶段都有暗坑:

  1. Tool definition 注入:依赖 chat template,缺特殊 token 直接挂
  2. 流式解析:增量 JSON parse,不是 JSON.parse
  3. 强制调用:模型会编造参数,需要兜底 no-op tool
  4. 结果注入:截断策略决定信息质量
  5. Parser 匹配:开源模型格式各异,parser 不对不报错但无结果
  6. 递归控制:必须有硬性终止条件
  7. 超时处理:把错误作为 tool result 返回,而非直接抛异常

把这些坑填了,Function Calling 才算从 Demo 级进化到生产级。

Continuous Batching 深度解析:为什么 LLM 推理不能用传统 Request Batching

· 阅读需 12 分钟

如果你做过 Web 后端,对"请求合并批量处理再返回"应该不陌生。但在 LLM 推理领域,这个模式完全行不通——传统静态 batching 的吞吐只有 Continuous Batching 的 十分之一不到。本文从后端工程师的视角,拆解这两种 batching 的根本差异。