数据截至 2025-07-20(上游 commit 97b9856e49f2)

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 Call	`SingleLLMCallNode`	一次模型调用,吐出结构化 JSON
Agent	`AgentNode`(继承前者)	带工具的 agent,循环调用工具直到收敛
Loop	`ForLoopNode`(继承 `BaseLoopSubworkflowNode`)	把一整个子工作流反复跑 N 次

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

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

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

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

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

事前:用 litellm 的 response_format 逼模型走结构化输出(json schema 模式)。
事后:模型还是不听话时,用 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 + user	`nodes/llm/_utils.py:69-83`
`generate_text`	litellm 统一调用 + 结构化输出协商	`nodes/llm/_utils.py:231`
`repair_json`	脏 JSON 暴力修复	`single_llm_call.py:20-93`
`json_schema_to_model`	schema → 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	触发关键字	面向用户的话
`overloaded`	`VertexAIError` + `The model is overloaded`	模型过载,稍后重试
`rate_limit`	`rate limit`	限流,过几分钟再试
`context_length`	`context length` / `maximum token`	输入太长超上下文窗口
`auth`	`invalid api key` / `authentication`	API key 有问题
`service_unavailable`	`bad 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_schema 是 BaseNode 的属性(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_calls 用 asyncio.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.py`	`SingleLLMCallNode` / `SingleLLMCallNode.run` / `SingleLLMCallNode.setup`
脏 JSON 修复	`backend/pyspur/nodes/llm/single_llm_call.py`	`repair_json`
litellm 调用 + 结构化协商	`backend/pyspur/nodes/llm/_utils.py`	`generate_text`
消息拼装	`backend/pyspur/nodes/llm/_utils.py`	`create_messages`
schema → 模型	`backend/pyspur/utils/pydantic_utils.py`	`json_schema_to_model`
Agent 节点	`backend/pyspur/nodes/llm/agent.py`	`AgentNode` / `AgentNode.run` / `AgentNode._call_tool` / `AgentNode.execute_parallel_tool_calls`
节点即工具	`backend/pyspur/nodes/base.py`	`BaseNode.function_schema` / `BaseNode.call_as_tool`
循环基类	`backend/pyspur/nodes/loops/base_loop_subworkflow_node.py`	`BaseLoopSubworkflowNode` / `run_iteration` / `stopping_condition`
For 循环	`backend/pyspur/nodes/loops/for_loop_node.py`	`ForLoopNode`
task 落库	`backend/pyspur/execution/task_recorder.py`	`TaskRecorder` / `create_task` / `update_task`
落库调用点	`backend/pyspur/execution/workflow_executor.py`	`WorkflowExecutor.is_downstream_of_pause`

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

1.1 一句话​

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

1.3 最小使用示意​

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

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

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

3.2 渲染:jinja2 拼 system / user 消息​

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

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

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

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

4. AgentNode:让节点长出手脚​

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

4.2 迭代式工具调用循环​

4.3 工具怎么被真正执行​

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

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

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

5.2 loop_history:循环的「记忆」​

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

5.4 动态拼最终输出模型​

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

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

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

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

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

8. 边界与局限​

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

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