LLM 编排框架基础

ToLO 发生在 LLM 输出离开模型之后。模型返回 token 只是第一步;真正的安全边界出现在框架把这些 token 包装成消息对象、解析成结构化字段、交给 agent 决策、传入工具或工作流执行环境时。

这一章不是 LLM 入门,也不是框架使用教程。它只关注与 ToLO 判断有关的组件:哪些组件产生 S_LLM source、哪些组件改变输出语义、哪些组件把输出推进危险 sink。

读完本章,你应当能拿到任意 LLM 应用代码,5 分钟内画出它的 source → transform → sink 链路。

这一章给你什么

你将能做到	用到的内容
在 OpenAI / Anthropic / LangChain / LlamaIndex 任一框架代码里指出 S_LLM 字段	§“跨框架的同构” + §“S_LLM 字段速查”
用一段最小代码示范”模型输出 → tool 调用”全过程	§“三个最小例子”
解释 PromptTemplate / Retriever / Parser / Agent / Tool / Workflow / Sandbox 各自的 ToLO 关注点	LLM Application Stack
区分”结构化输出” 与”安全输出”	§“常见误解”

如果你已经熟 LangChain agent 写法,可以直接跳到 LLM Application Stack。

你需要先知道什么

只需要 先修知识 §1-§4 涵盖的:

• LLM 输出由 token 组成,模型本身只生成 token,不”做事”。
• 让 LLM “做事” 的是框架代码:把输出解析成指令,转译成函数调用 / SQL / shell / HTTP / 文件操作。
• 结构化输出(JSON schema / Pydantic / function calling)只保证形状,不保证内容。

如果上面任一项不熟,先回去补。

主流框架的同构

不同框架在 ToLO 视角下做着同一类事。一张速查表:

角色	LangChain	OpenAI SDK	Anthropic SDK	LlamaIndex	MCP
模型客户端	ChatOpenAI, ChatAnthropic, …	OpenAI()	Anthropic()	OpenAI() 同	client 实现
输入消息	[SystemMessage, HumanMessage, ToolMessage]	messages=[{"role":...,"content":...}]	messages=[{"role":...,"content":...}]	同 OpenAI SDK	request params
输出	AIMessage	response.choices[0].message	response.content[i]	Response	response result
输出字段	.content, .tool_calls[i].args	.message.content, .message.tool_calls[i].function.arguments	.content[i].text, .content[i].input	.response, .source_nodes	result.content[]
工具描述	@tool / Tool(name, func)	tools=[{type:"function",...}]	tools=[{name, input_schema, ...}]	FunctionTool	server 暴露的 tools/list
RAG 文档	Document.page_content	(自管)	(自管)	NodeWithScore.node.text	server 暴露的 resources

关键观察:字段名不同,但类型角色都一样。AIMessage.content 和 response.choices[0].message.content 在 ToLO 里没有任何区别 —— 都是 S_LLM^direct。一条 source spec 可以同时覆盖多个框架。

三个最小例子

例 1:OpenAI SDK 裸函数调用(最底层)

import json
from openai import OpenAI
client = OpenAI()

tools = [{
    "type": "function",
    "function": {
        "name": "read_file",
        "description": "读取服务器上的文本文件",
        "parameters": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
            },
            "required": ["path"],
        },
    },
}]

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "你是一个文件助手。"},
        {"role": "user",   "content": "把 /etc/hostname 的内容给我"},
    ],
    tools=tools,
)

message = response.choices[0].message      # 整个消息对象
content = message.content                  # ← S_LLM^direct (可能是 None,模型只调工具时)
tool_call = message.tool_calls[0]
args_json = tool_call.function.arguments   # ← S_LLM^framework (JSON 字符串)
args = json.loads(args_json)               # ← S_LLM^parsed (dict)

path = args["path"]                        # ← S_LLM^parsed (str 字段)
print(open(path).read())                   # ← ToLO-Path sink

四行 source 注释把整条链路标完了:模型输出 → JSON 解析 → 路径字段 → open 读文件。任何一行都是 untrusted。

例 2:LangChain agent(高层封装)

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate

@tool
def read_file(path: str) -> str:
    """读取服务器上的文本文件"""
    return open(path).read()           # ← ToLO-Path sink

llm = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个文件助手。"),
    ("human",  "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])
agent = create_tool_calling_agent(llm, [read_file], prompt)
executor = AgentExecutor(agent=agent, tools=[read_file])
executor.invoke({"input": "把 /etc/hostname 的内容给我"})

表面没看到 source/sink,但里面隐藏着同样的链路:

• AgentExecutor 内部:AIMessage.tool_calls[0]["args"] ← S_LLM^framework
• LangChain 把 args 喂进 read_file(path=...)
• open(path) ← ToLO-Path sink

LangChain 把链路藏在 AgentExecutor 里,但链路本身和例 1 一样。审计 LangChain 代码时,要心里清楚 @tool 函数的参数都是 LLM 输出,而非”内部值”。

例 3:LlamaIndex RAG(把检索加进来)

from llama_index.core import VectorStoreIndex, Document
from llama_index.llms.openai import OpenAI as LIOpenAI

docs = [
    Document(text="HR 政策:报销需 7 天内提交。"),
    Document(text=open("/path/to/some/external.md").read()),  # ← 可能被投毒
]
index = VectorStoreIndex.from_documents(docs)
query_engine = index.as_query_engine(llm=LIOpenAI(model="gpt-4o-mini"))

response = query_engine.query("我们的报销政策是什么?")
print(response.response)                  # ← S_LLM^direct
for node in response.source_nodes:        # ← S_LLM^rag
    print(node.node.text)

这里:

• Document(text=...) 来自外部文件,内容可控 → C3 通道入口。
• query_engine.query(...) 内部:先做 embedding + 相似度检索,把命中 doc 拼进 prompt,发给 LLM。
• response.response 是 LLM 输出,受 doc 内容影响。
• response.source_nodes[i].node.text 是检索回来的 doc 内容,是 S_LLM^rag。

只要应用拿 response.response 喂任何危险 sink(比如 eval、open、requests.get),就构成 ToLO。

S_LLM 字段速查(跨框架版)

读代码时,看到下面任一字段访问,立刻把它当 S_LLM:

S_LLM^direct      ← 模型直接生成的文本
  • OpenAI:        response.choices[0].message.content
  • Anthropic:     response.content[i].text
  • LangChain:     AIMessage.content
  • LlamaIndex:    Response.response
  • 任何 streaming: chunk.delta.content / chunk.choices[0].delta.content

S_LLM^framework   ← 框架把模型输出包装成的对象字段
  • OpenAI:        response.choices[0].message.tool_calls[i].function.arguments (str)
  • Anthropic:     response.content[i].input (dict, tool_use block)
  • LangChain:     AIMessage.tool_calls[i]["args"]
  • LangChain:     AgentAction.tool_input
  • LangChain:     AgentFinish.return_values
  • MCP:           tools/call 返回的 content[i].text

S_LLM^parsed      ← OutputParser / json.loads / regex 提取后的字段
  • LangChain:     PydanticOutputParser 输出实例的字段
  • LangChain:     JsonOutputParser 输出 dict 的字段
  • 任何 json.loads(message.content)["..."]
  • 任何 re.search(pattern, message.content).group(1)

S_LLM^structured  ← structured output / function calling 强类型字段
  • OpenAI:        client.beta.chat.completions.parse(response_format=...)
  • LangChain:     llm.with_structured_output(MyPydantic).invoke(...)
  • Anthropic:     response.content[i].input(tool_use block)

S_LLM^rag         ← RAG 检索回的文档内容
  • LangChain:     Document.page_content, Document.metadata
  • LlamaIndex:    NodeWithScore.node.text, .metadata
  • 任何 vector_store.similarity_search(...) 的返回值

打印一下:这五个子集合在一起就是 ToLO 的 source 集。本站后续所有规则、所有案例,都在追踪这些字段最终去了哪里。

三个组件,三种不同的安全错觉

LLM 框架里有三个组件特别容易制造错觉,审计时要小心。

错觉 1:OutputParser 让数据”看起来已经清洗”

from langchain_core.output_parsers import JsonOutputParser

raw = AIMessage(content='{"sql": "SELECT * FROM users"}')
parsed = JsonOutputParser().invoke(raw)   # ← dict({"sql": "SELECT * FROM users"})
db.execute(parsed["sql"])                  # ← ToLO-SQL

JsonOutputParser 只是 json.loads 的包装。它只验证 JSON 是合法的,不验证字段内容是否安全。

错觉 2:Structured output / function calling 让数据”看起来强类型”

from pydantic import BaseModel

class SqlQuery(BaseModel):
    sql: str

result: SqlQuery = client.beta.chat.completions.parse(
    model="gpt-4o-mini",
    messages=[...],
    response_format=SqlQuery,
)
db.execute(result.sql)   # ← ToLO-SQL 仍然成立

Pydantic 保证 result.sql 是 str 类型,不是 None、不是 int、不是其他形状。但 str 可以是任何字符串,包括 "DROP TABLE users; --"。

错觉 3:Tool wrapper 让数据”看起来被框架检查过”

@tool
def query_db(sql: str) -> str:
    """执行 SQL 查询"""
    return str(db.execute(sql).fetchall())   # ← ToLO-SQL

@tool 装饰器只做 schema 暴露,告诉 LLM “这个工具叫 query_db,参数是 sql:str”。它不检查参数内容、不限制可执行 SQL 形状、不切断危险操作。

审计 LangChain / LlamaIndex 代码时,@tool 函数体内的危险调用就是 sink,参数就是 source。

常见误解

误解 1:“框架默认包含安全过滤”

不会。LangChain / LlamaIndex / OpenAI SDK 都不会默认对 LLM 输出做内容安全过滤。它们提供组件,组件之间怎么连是开发者的选择。

误解 2:“用了 agent framework,我就不用考虑 sink”

错。Agent framework 把 sink 藏得更深(在 @tool 函数体里),但 sink 仍然在。审计时要打开每个 tool 看里面。

误解 3:“OpenAI 的 function calling 比自己写 prompt 安全”

只能说安全一点点。function calling 保证模型输出的参数符合 schema(形状),但参数内容仍可被攻击者影响。

误解 4:“我用了 MCP 标准协议,所以安全”

错。MCP 只规定通信协议(JSON-RPC 格式),不规定 server 实现是否安全。MCP server 是第三方代码,它的返回值默认 untrusted(C4 通道)。

误解 5:“我让模型生成代码后,用 sandbox 跑就行”

部分对。Sandbox(如 RestrictedPython、Docker、gVisor)能限制 sink 影响范围,确实属于 C_SAFE^capability。但要看 sandbox 真的封住了什么:

• 文件系统隔离了吗?
• 网络访问限制了吗?
• 可访问的 Python 内置函数收窄了吗?
• 内存 / CPU / 时间 quota 有吗?

裸装 RestrictedPython 不够,要看具体配置。

一个综合审计练习

考虑这段稍复杂的 LangChain 代码,标出所有 S_LLM 字段和潜在 sink:

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
import subprocess, requests

vectorstore = Chroma(persist_directory="./db", embedding_function=OpenAIEmbeddings())

@tool
def search_docs(query: str) -> str:
    """在知识库里搜索文档"""
    docs = vectorstore.similarity_search(query, k=3)
    return "\n".join(d.page_content for d in docs)     # ← ❶ ❷

@tool
def fetch_url(url: str) -> str:
    """获取 URL 内容"""
    return requests.get(url, timeout=5).text            # ← ❸

@tool
def run_diagnostic(cmd: str) -> str:
    """运行诊断命令"""
    return subprocess.run(cmd, shell=True, capture_output=True).stdout.decode()   # ← ❹

llm = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是 SRE 助手,可以查文档、抓 URL、运行诊断。"),
    ("human",  "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])
agent = create_tool_calling_agent(llm, [search_docs, fetch_url, run_diagnostic], prompt)
executor = AgentExecutor(agent=agent, tools=[search_docs, fetch_url, run_diagnostic])
result = executor.invoke({"input": user_input})
return result["output"]                                  # ← ❺

标注答案:

• ❶ d.page_content ← S_LLM^rag(C3 通道入口)。如果 Chroma 数据库里被写入投毒文档,这里就成 untrusted。
• ❷ tool 返回值(search_docs 的 return)会作为 ToolMessage 拼回 LLM 上下文 → 影响下一轮 AIMessage.content。也就是 S_LLM^direct 被污染。
• ❸ requests.get(url, ...):url 来自 S_LLM^framework → ToLO-SSRF sink。
• ❹ subprocess.run(cmd, shell=True, ...):cmd 来自 S_LLM^framework → ToLO-Shell sink,且 shell=True 让攻击面最大。
• ❺ result["output"] ← S_LLM^direct。如果调用方拿它继续做事(比如展示到 HTML / 存到日志 / 喂给另一个解析器),要继续追踪。

三个 sink 没有任何防御。修复方向(详见 Defensive Patterns):

• fetch_url:URL allowlist + 内网地址 block + scheme 限制 (C_SAFE^allowlist)。
• run_diagnostic:根本就不应该让 LLM 决定 shell 命令。改成枚举命令(Literal["ping", "traceroute"] + 固定参数模板)。shell=True 永远不要用,改成 subprocess.run([...]) list 形式 (C_SAFE^parameterized)。
• search_docs:RAG 文档需要来源审核 / 签名 + 在拼入 prompt 前考虑标记可信级别。

读完检查

继续往下读前,确认你能:

• 写出 OpenAI / Anthropic / LangChain / LlamaIndex 各自至少一个 S_LLM^direct 字段路径。
• 写出至少 3 个 S_LLM^framework 字段。
• 说明为什么 JsonOutputParser 不是 sanitizer。
• 说明 @tool 装饰器不会做安全检查。
• 拿到任意 LangChain agent 代码,5 分钟内圈出所有 sink。

下一步阅读

• LLM Application Stack:把 8 个组件按数据流顺序展开,每个组件都给 ToLO 关注点。
• Core ToLO Patterns:七子类如何在 sink 维度分化。
• Trust Boundaries:五个攻击者通道和被保护对象。