跳到主要内容

LLM 与 Agent 节点:结构化输出、工具调用与循环

30 秒导读: 前面几章讲的是「DAG 怎么被描述、怎么被调度、怎么暂停恢复」。这一章落到真正干活的那几个节点——把一个大模型塞进 DAG 的一个格子里,让它:① 输出能被下游节点直接吃的结构化 JSON;② 当 agent 一样反复调用工具;③ 被 Loop 节点反复跑。最后收尾讲这一切怎么落库(TaskRecorder)。

本章是 PySpur 文档的最后一章,假设你已经读过:

  • 03-node-system.md —— BaseNode 抽象、动态 Pydantic 模型、工厂注册。本章的三个节点都是 BaseNode 的子类。
  • 02-executor.md —— 拓扑调度与失败传播。本章的 Loop 节点内部嵌了一个 WorkflowExecutor
  • 04-human-in-loop.md —— 暂停/恢复状态机。本章末尾的 TaskRecorder 就是它反复用到的落库机制。

框架层与调度层这里不重复,只讲具体节点实现 + task 落库


1. 这是什么(先建直觉)

1.1 一句话

PySpur 里,「调用大模型」不是散落各处的 openai.chat(...),而是三个节点类:

节点干什么
Single LLM CallSingleLLMCallNode一次模型调用,吐出结构化 JSON
AgentAgentNode(继承前者)带工具的 agent,循环调用工具直到收敛
LoopForLoopNode(继承 BaseLoopSubworkflowNode)把一整个子工作流反复跑 N 次

它们都是 DAG 里的普通格子,输入输出都走第 03 章那套动态 Pydantic 模型。

1.2 为什么难 —— 「把话变成数据」

一个大模型只会吐文本。可 DAG 的下游节点要的是字段:output.answeroutput.score。中间隔着一道鸿沟:

模型说的: "当然可以!结果是 325。```json {"result": 325} ```"
下游要的: output.result == 325 ← 一个干净的、schema 对得上的对象

这道鸿沟就是本章工程含量最高的地方。PySpur 的做法是双保险:

  1. 事前:用 litellm 的 response_format 逼模型走结构化输出(json schema 模式)。
  2. 事后:模型还是不听话时,用 repair_json 暴力修它吐出来的脏 JSON。

1.3 最小使用示意

下面这段来自 single_llm_call.py 文件底部的自测,直观感受一个 LLM 节点长什么样(真实代码,backend/pyspur/nodes/llm/single_llm_call.py:369-403):

# 示意:节选自源码自测,展示节点配置形态
simple_llm_node = SingleLLMCallNode(
name="WeatherBot",
config=SingleLLMCallNodeConfig(
llm_info=ModelInfo(model=LLMModels.GPT_4O, temperature=0.4, max_tokens=100),
system_message="You are a helpful assistant.",
user_message="Hello, my name is {{ name }}. I want to ask: {{ question }}", # jinja2 模板
output_json_schema=json.dumps({ # ← 逼模型按这个 schema 输出
"type": "object",
"properties": {"answer": {"type": "string"}, "name_of_user": {"type": "string"}},
"required": ["answer", "name_of_user"],
}),
),
)

重点看两处:user_message 里的 {{ name }}jinja2 占位符(运行时用输入填上),output_json_schema输出契约(运行时用来生成 Pydantic 校验模型)。


2. 顶层全景(一次 LLM 调用怎么走完)

先看 SingleLLMCallNode.run() 的主线(single_llm_call.py:171-359),它是另外两个节点的地基:

输入(Pydantic model)
│ input.model_dump() → raw_input_dict

① 渲染消息 Template(system_message).render(dict) ← jinja2
Template(user_message).render(**dict)
│ (user_message 为空 → 直接 dump 整个输入为 JSON)

② 注入历史 enable_message_history → 从输入取 message_history 列表
│ create_messages(system, user, few_shot, history)

③ 调模型 generate_text(..., json_mode=True,
│ output_json_schema=...) ← litellm

④ 解析 json.loads(content)
│ └─失败→ repair_json(content) 再 loads ← 事后兜底
│ └─还失败→ 抛结构化 parsing_error

⑤ 校验 self.output_model.model_validate(dict) ← 动态生成的模型

输出(Pydantic model)

四个部件、各在哪个文件:

部件干什么位置
jinja2 Template把输入填进 system/user 消息single_llm_call.py:176,183
create_messages拼 system + few-shot + history + usernodes/llm/_utils.py:69-83
generate_textlitellm 统一调用 + 结构化输出协商nodes/llm/_utils.py:231
repair_json脏 JSON 暴力修复single_llm_call.py:20-93
json_schema_to_modelschema → Pydantic 输出模型utils/pydantic_utils.py:27

围绕这条主线,包了一层分层错误分类(把 litellm 各种异常翻译成结构化错误 JSON),下面逐个拆。


3. SingleLLMCallNode:一次调用的全部工程细节

3.1 setup():把 schema 变成输出模型

节点被工厂创建后会调 setup()。这里做一件关键事:如果配置了 output_json_schema,就动态生成一个 Pydantic 输出模型(single_llm_call.py:162-169):

def setup(self) -> None:
super().setup()
if self.config.output_json_schema:
self.output_model = json_schema_to_model(
json.loads(self.config.output_json_schema),
self.name,
SingleLLMCallNodeOutput, # 以它为基类
)

这就是第 03 章「动态 Pydantic 模型」在 LLM 节点上的具体体现:用户在 UI 里画的输出 schema,变成一个真正的类,后面 model_validate 就用它。

3.2 渲染:jinja2 拼 system / user 消息

run() 开头把输入整个 dump 成字典,再用 jinja2 渲染(single_llm_call.py:172-187):

raw_input_dict = input.model_dump()
system_message = Template(self.config.system_message).render(raw_input_dict)
if not self.config.user_message.strip():
user_message = json.dumps(raw_input_dict, indent=2) # 没写 user_message → 把整个输入丢给模型
else:
user_message = Template(self.config.user_message).render(**raw_input_dict)

一个小设计:user_message 留空时,直接把整个输入字典序列化当用户消息。这让「把上游所有输出一股脑喂给模型」成为零配置的默认行为。

3.3 消息历史注入:让节点能对话

enable_message_history 打开时,节点会从输入里的某个变量取出历史消息列表(single_llm_call.py:189-207):

history_var = self.config.message_history_variable
if "." in history_var: # 支持 "input_node.message_history" 这种嵌套引用
history = get_nested_field(history_var, input)
else:
history = raw_input_dict.get(history_var)
assert isinstance(history, list) or history is None

历史随后被 create_messages 插在 few-shot 之后、当前 user 消息之前(_utils.py:80-82):system → few_shot → history → user。这是标准的对话上下文顺序。

3.4 结构化输出:litellm 怎么被逼吐 JSON

真正逼模型走结构化的逻辑在 generate_text(_utils.py:231,json_mode=True 由节点传入)。它按模型能力分三档降级(_utils.py:294-339):

模型支持 response_schema(或 anthropic) → response_format = json_schema(strict) 最强,schema 级约束
│否
模型支持 response_format → response_format = json_object 弱,只保证是 JSON
│ + 把 schema 塞进 system 提示词里(靠提示约束)
│否
完全不支持 → 不加 response_format,纯靠事后 repair

关键片段(_utils.py:312-339):

if litellm.supports_response_schema(model=model_name, ...) or model_name.startswith("anthropic"):
kwargs["response_format"] = {"type": "json_schema", "json_schema": output_json_schema}
else:
kwargs["response_format"] = {"type": "json_object"}
# 模型不认 schema,就把 schema 拼进 system 消息,靠提示词约束
system_message["content"] += (
"\nYou must respond with valid JSON only. ... "
"The JSON Object must adhere to this schema: " + schema_for_prompt
)

另外两处细节:schema 会先被 sanitize_json_schema 清洗,再统一加 additionalProperties = False(_utils.py:306-309);litellm 全局开了 drop_params = True(_utils.py:29),不支持某参数的模型会自动丢掉而非报错。

3.5 repair_json:脏 JSON 的暴力修复(本章招牌)

模型仍可能吐出带围栏、带单引号、缺逗号、前后有废话的伪 JSON。repair_json(single_llm_call.py:20-93)是一条流水线式兜底,按顺序拧:

步骤修什么代码行
去 XML/思考标签</invoke><thinking> 等混入片段:32
去 markdown 围栏```json 和结尾 ```:35
截取 JSON 主体正则抠出第一个 {...}:38-40
单引号转双引号先占位保护已合法的双引号串,再把剩余 '",最后还原:44-63
去尾逗号,} / ,] 里的多余逗号:66
补缺失逗号}{"[" 之间插逗号:69
给裸键加引号{key:{"key"::72
截首尾大括号find("{")rfind("}") 兜底再抠一遍:82-88

「单引号转双引号」那步最见功力——不能无脑替换,否则会破坏字符串内部合法的双引号内容,所以先把合法双引号串换成占位符 PLACEHOLDER{n}、替换完再还原(single_llm_call.py:44-63):

# 示意:先保护、后替换、再还原,避免误伤字符串内部的引号
quoted_strings = {}
repaired = re.sub(r'"[^"\\]*(?:\\.[^"\\]*)*"', replace_quoted, repaired) # 双引号串 → 占位符
repaired = repaired.replace("'", '"') # 剩下的单引号安全转
for key, value in quoted_strings.items():
repaired = repaired.replace(key, value) # 还原

调用点在 run() 里,是标准的「先试正常解析、失败再修」(single_llm_call.py:258-263):

try:
assistant_message_dict = json.loads(assistant_message_content)
except Exception:
repaired_str = repair_json(assistant_message_content) # 兜底
assistant_message_dict = json.loads(repaired_str)

坑: repair_json 是启发式的,修不好会连锁抛出结构化 parsing_error(见 3.6)。它兜的是「模型格式小毛病」,不是「模型答非所问」。

3.6 分层错误分类:把 litellm 异常翻译成结构化错误 JSON

run() 的第二个亮点是错误不外泄成原始堆栈,而是分类成结构化 JSON。分三类:

① 解析类(parsing_error)——repair_json 也救不回来(single_llm_call.py:264-280),错误体带上 assistant_message_str 原文,方便排查。

② 模型服务类(model_provider_error)——只要错误串里含 litellm,就按关键字细分(single_llm_call.py:281-329):

if "litellm" in error_str.lower():
provider = model_name.split("/")[0] if "/" in model_name else "unknown"
if "VertexAIError" in error_str and "The model is overloaded" in error_str:
error_type = "overloaded"
elif "rate limit" in error_str.lower(): error_type = "rate_limit"
elif "context length" in error_str.lower(): error_type = "context_length"
elif "invalid api key" in error_str.lower(): error_type = "auth"
elif "bad gateway" in error_str.lower(): error_type = "service_unavailable"
raise Exception(json.dumps({"type": "model_provider_error", "provider": provider, ...}))

覆盖的错误类型:

error_type触发关键字面向用户的话
overloadedVertexAIError + The model is overloaded模型过载,稍后重试
rate_limitrate limit限流,过几分钟再试
context_lengthcontext length / maximum token输入太长超上下文窗口
authinvalid api key / authenticationAPI key 有问题
service_unavailablebad gateway / 503服务暂时不可用

③ schema 不符类(invalid_json_format)——JSON 合法但 model_validate 不过(single_llm_call.py:332-359),错误体里同时带上原始响应尝试校验的对象,方便定位是模型漏字段还是类型错。

这套分类的价值:上游 UI / API 拿到的是type 字段的结构化错误,能针对性提示或重试,而不是一坨 traceback。


4. AgentNode:让节点长出手脚

AgentNode 继承 SingleLLMCallNode(agent.py:59),复用了上面全部的渲染/解析/错误分类,只加两样:工具迭代循环

4.1 工具从哪来 —— 子工作流里的普通节点

Agent 的配置里挂了一个 subworkflow(agent.py:40-45)。子工作流里的每个节点,就是一个工具setup() 把它们转成三份数据(agent.py:87-117):

for tool in tools:
self.tools_dict[tool.title.lower()] = tool # 名字 → 节点定义(查表用)
self.tools_instances[tool.title.lower()] = NodeFactory.create_node(...) # 名字 → 节点实例
self.tools_schemas.append(tool_node_instance.function_schema) # 喂给 LLM 的函数 schema

function_schemaBaseNode 的属性(nodes/base.py:263-316):它把节点的 config 模型 schema 翻成 OpenAI 风格的 function schema——节点的配置字段变成函数参数,docstring 变成函数描述:

function_schema = {
"type": "function",
"function": {
"name": self.name,
"description": description, # 来自类 docstring
"parameters": {"type": "object", "properties": properties, "required": required},
},
}

这是很漂亮的复用:任何一个 PySpur 节点,天然就能当 agent 的工具,不需要单独写工具封装。

4.2 迭代式工具调用循环

AgentNode.run() 的核心是一个 while 循环,最多转 max_iterations 次(默认 10,agent.py:262-306):

┌─────────────────────────────────────────────┐
│ while num_iterations < max_iterations: │
│ │
│ generate_text(messages, tools=schemas) │ ← 带工具调模型
│ │ │
│ ├─ 有 tool_calls? ──是──► 并行执行工具 │
│ │ 结果回灌 messages
│ │ continue ────┤ 回到循环顶
│ │ │
│ └─ 是 assistant 文本? ─是─► 记录、break │
└─────────────────────────────────────────────┘

▼ 解析 model_response(json.loads / repair_json)
输出对象

关键片段(agent.py:262-293):

while num_iterations < self.config.max_iterations:
message_response = await generate_text(messages=messages, ..., tools=self.tools_schemas)
messages.append({"role": message_response.role,
"content": str(message_response.content),
"tool_calls": message_response.tool_calls})
num_iterations += 1
if message_response.tool_calls: # 模型想调工具
tool_responses = await self.execute_parallel_tool_calls(message_response.tool_calls)
messages.extend(cast(List[Dict[str, str]], tool_responses)) # 工具结果回灌
continue # 再转一圈
elif message_response.role == "assistant" and message_response.content is not None:
model_response = str(message_response.content) # 模型给了最终答案
break

4.3 工具怎么被真正执行

一轮里可能有多个工具调用,execute_parallel_tool_callsasyncio.gather 并行跑(agent.py:165-185)。单个工具的执行落在 _call_tool(agent.py:141-163):

tool_node = self.tools_dict.get(tool_name.lower()) # 按模型给的名字查表
tool_node_instance = NodeFactory.create_node(...) # 现造一个实例
tool_args = json.loads(tool_args) # 模型给的是 JSON 字符串参数
return await tool_node_instance.call_as_tool(arguments=tool_args) # 用参数当 config+input 跑

call_as_tool(nodes/base.py:318-333)有个巧思:它把模型给的 arguments 同时当作 config 和 input——config_model.model_validate(arguments) 造配置,input_model.model_validate(arguments) 造输入,再 run。因为工具的「参数」在 function_schema 里本就是从 config 字段翻出来的,这里正好对上。

结果被包成 ChatCompletionToolMessage(role=tool)回灌进 messages,下一轮模型就能看到工具输出继续推理(agent.py:171-179)。

4.4 收尾:把工具轨迹塞进输出

循环结束后,同样用 json.loads / repair_json 解析最终文本(复用父类逻辑,agent.py:357-379),然后把整段对话里的工具消息当轨迹塞进输出的 tool_calls 字段(agent.py:381-383):

assistant_message_dict["tool_calls"] = [
str(m) for m in messages if m["role"] == "tool" or "tool_calls" in m
]
assistant_message = self.output_model.model_validate(assistant_message_dict)

这样下游节点既拿到答案,也拿到「它调了哪些工具」的记录。

边界: 若循环跑满 max_iterations 模型还在调工具,model_response 会一直是空串 "",后续解析空串会走进 repair_json(空输入返回 "{}"),最终多半在 schema 校验时报 invalid_json_format。Agent 没有显式的「超轮数」错误类型 (inferred,基于 agent.py:259-306 无 else 分支处理满轮情形)。


5. Loop:把一整个子工作流反复跑

对应 README 的 🔄 Loops: Iterative tool calling with memory 特性(README.md:59,118)。实现基类是 BaseLoopSubworkflowNode(nodes/loops/base_loop_subworkflow_node.py),具体子类 ForLoopNode 只补一个停止条件。

5.1 思路 —— 内嵌一个执行器,反复跑

和第 02 章的 WorkflowExecutor 关系是套娃:Loop 节点本身是外层 DAG 的一个格子,但它内部每转一圈就新建一个 WorkflowExecutor 跑子工作流(base_loop_subworkflow_node.py:56-79):

外层 DAG
└─[Loop 节点].run()
while not stopping_condition(current_input): ← 5.3
run_iteration:
iteration_input = {**input, "loop_history": self.loop_outputs} ← 注入历史 5.2
executor = WorkflowExecutor(subworkflow) ← 每轮新建
outputs = await executor.run(iteration_input)
_update_loop_outputs(outputs) ← 累积历史
current_input.update(iteration_output) ← 上轮输出喂下轮

5.2 loop_history:循环的「记忆」

每轮开始前,把到目前为止所有轮的输出作为 loop_history 注入子工作流输入(base_loop_subworkflow_node.py:62):

iteration_input = {**input, "loop_history": self.loop_outputs}

loop_outputs{node_id: [每轮该节点的输出, ...]}。子工作流里的节点就能读到「上几轮我算到哪了」——这就是 README 说的 with memory_update_loop_outputs(:39-49)负责按轮追加,并刻意跳过 loop_history 字段本身,避免历史套历史无限膨胀(:43-44)。

for_loop_node.py:64-72 的自测演示了子节点怎么用它:读 loop_history.get('increment', []) 累加出一个 running_total

5.3 停止条件:模板方法模式

基类把 stopping_condition 声明为 @abstractmethod(:51-54),run()while not stopping_condition(...)(:86-89)。ForLoopNode 只需填一行(for_loop_node.py:30-32):

async def stopping_condition(self, input: Dict[str, Any]) -> bool:
return self.iteration >= self.config.num_iterations # 跑够 N 次就停

换个停止条件(比如「直到某字段为 true」)就是另一种 Loop 节点,基类完全复用。

5.4 动态拼最终输出模型

循环结束后要产出一个符合外层期望的输出。Loop 节点不预先知道自己输出什么字段——它照抄子工作流里 OutputNode 的输出模型(base_loop_subworkflow_node.py:93-108):

output_node = next(n for _id, n in self._executor.node_instances.items()
if issubclass(n.__class__, OutputNode))
self.output_model = create_model( # 运行时现造输出模型
f"{self.name}",
**{name: (field, ...) for name, field in output_node.output_model.model_fields.items()},
__base__=BaseLoopSubworkflowNodeOutput, ...
)
return self.output_model.model_validate(current_input)

又一次「动态 Pydantic 模型」的运用(呼应 03-node-system.md):Loop 的输出 schema 是从内嵌 OutputNode 复制来的,而不是写死的。


6. 收尾:TaskRecorder —— 这一切怎么落库

前面 02/04 章反复提到「节点跑一次就落一条 task 记录」,机制集中在 TaskRecorder(execution/task_recorder.py)。它由执行器持有(workflow_executor.py:57),节点开跑前 create_task、跑完 update_task(workflow_executor.py:171,307)。

6.1 为什么需要它 —— 重跑/恢复要去重

一个 run 可能被多次执行(尤其 04 章的暂停/恢复:一个工作流会跑好几趟)。同一个 node_id 因此可能在库里留下多条 task。构造函数要从这堆里挑出唯一一条最该信的(task_recorder.py:34-80),按状态优先级:

COMPLETED > PAUSED > RUNNING > PENDING > FAILED > CANCELED
(已完成的最权威,COMPLETED 里取 end_time 最新的那条)

代码就是一串顺序 if(task_recorder.py:42-77),先 COMPLETED、再 PAUSED……第一个命中即用。这保证恢复时已完成的节点不会被重跑

6.2 create_task:不覆盖已定局的记录

create_task(task_recorder.py:82-152)的核心原则:COMPLETED / PAUSED / RUNNING 的 task 不重建(:91-102),只在缺输入时补一下;只有 PENDING / FAILED / CANCELED 才会被「复活」成 RUNNING 重跑(:104-113)。内存缓存里没有时还会回查数据库,同样保护 COMPLETED / PAUSED(:115-140)。

6.3 update_task:暂停下游的特殊处理

update_task(task_recorder.py:154-198)有个和 04 章直接咬合的关键分支(:171-173):

# 若本节点在某个「已暂停」节点的下游,别把它标成失败/取消,而是标 PENDING
if is_downstream_of_pause and status in [TaskStatus.FAILED, TaskStatus.CANCELED]:
status = TaskStatus.PENDING
error = None # 顺手清掉误报的错误

语义: 上游有节点停在 human-in-the-loop 断点,那它下游那些「没跑成」的节点不是真失败,只是还没轮到——所以标 PENDING(等恢复后再跑)而非 FAILED。执行器在检测到下游于暂停点时正是带着 is_downstream_of_pause=True 调它的(workflow_executor.py:326-330,418-422)。这就是 04 章「暂停不污染下游状态」在落库层的落点。

update_task 还负责把 subworkflow / subworkflow_output(Agent 和 Loop 的内层结果)序列化存下(task_recorder.py:185-195),供前端回看 agent 的工具轨迹和循环各轮输出。


7. 巧妙之处(带走的精华)

  • 双保险结构化输出:事前用 litellm response_format 按模型能力三档降级逼 JSON,事后用 repair_json 兜底——两头都堵。_utils.py:294-339 + single_llm_call.py:20-93
  • repair_json 的引号保护:先把合法双引号串换占位符再转单引号,避免误伤字符串内部——启发式修复里最容易写错的一步做对了。single_llm_call.py:44-63
  • 任何节点即工具:function_schema 把节点 config 直接翻成 function schema,call_as_tool 把模型给的参数同时当 config+input,零额外封装。nodes/base.py:263-333
  • 动态输出模型三处复用:LLM 节点从 output_json_schema 生成、Loop 节点从内嵌 OutputNode 复制——同一套 create_model / json_schema_to_model 机制贯穿。
  • 暂停下游标 PENDING 而非 FAILED:落库层用一个布尔参数就把「没跑成」和「真失败」区分开,让恢复干净。task_recorder.py:171-173

8. 边界与局限

  • repair_json 只治格式,不治内容:模型答非所问、漏必填字段,它救不回来,最终报 invalid_json_format(single_llm_call.py:332-359)。
  • Agent 满轮无专门错误:跑满 max_iterations 仍在调工具时,model_response 为空,靠后续 schema 校验间接失败,无「超轮数」这一错误类型 (inferred)。
  • 错误分类靠关键字匹配:"litellm" in error_str.lower() 和一串子串匹配(single_llm_call.py:285-316)对 litellm 的错误文案有隐性依赖,上游改文案可能漏判成 unknown
  • Loop 每轮新建 WorkflowExecutor:run_iteration 每次 WorkflowExecutor(...)(base_loop_subworkflow_node.py:65),轮数多时有重复构造开销;历史全存内存 loop_outputs,超长循环内存会涨。

9. 横向对比(同组其它章)

关注点本章落点去哪章看框架
节点抽象 / 动态模型 / 工厂三个节点都用到03-node-system.md
拓扑调度 / 并发 / 失败传播Loop 内嵌执行器复用它02-executor.md
暂停 / 恢复状态机TaskRecorder 的 is_downstream_of_pause 咬合它04-human-in-loop.md
DAG 数据模型工具 = 子工作流节点、Loop = 子工作流01-workflow-model.md

10. 代码地图(导航索引)

主题文件路径符号名
单次 LLM 节点backend/pyspur/nodes/llm/single_llm_call.pySingleLLMCallNode / SingleLLMCallNode.run / SingleLLMCallNode.setup
脏 JSON 修复backend/pyspur/nodes/llm/single_llm_call.pyrepair_json
litellm 调用 + 结构化协商backend/pyspur/nodes/llm/_utils.pygenerate_text
消息拼装backend/pyspur/nodes/llm/_utils.pycreate_messages
schema → 模型backend/pyspur/utils/pydantic_utils.pyjson_schema_to_model
Agent 节点backend/pyspur/nodes/llm/agent.pyAgentNode / AgentNode.run / AgentNode._call_tool / AgentNode.execute_parallel_tool_calls
节点即工具backend/pyspur/nodes/base.pyBaseNode.function_schema / BaseNode.call_as_tool
循环基类backend/pyspur/nodes/loops/base_loop_subworkflow_node.pyBaseLoopSubworkflowNode / run_iteration / stopping_condition
For 循环backend/pyspur/nodes/loops/for_loop_node.pyForLoopNode
task 落库backend/pyspur/execution/task_recorder.pyTaskRecorder / create_task / update_task
落库调用点backend/pyspur/execution/workflow_executor.pyWorkflowExecutor.is_downstream_of_pause