LLM Application Stack

典型 LLM 应用栈由 8 个组件按一条数据流串起来。本页一个一个讲:每个组件做什么、产生哪些 S_LLM 字段、可能落入哪个 ToLO 子类、审计时看什么。

读完后,任意陌生 LLM 应用代码,你都能按”输入 → 模型 → 解析 → 决策 → 执行 → 反馈”顺序拆成可分析的数据流。

这一节要建立的直觉

普通聊天应用的路径很短:

用户输入 → LLM → 展示回答

这种场景未必有 ToLO:输出虽然可能错误或有诱导性,但没有进入危险 sink。

编排框架的路径更长:

用户输入 → PromptTemplate → LLM → OutputParser → Agent → Tool → Workflow loop → Sandbox? → Sink

一旦最后几步连到数据库、shell、文件系统、网络请求或代码解释器,LLM 输出就不再只是文本,而是在影响程序行为。ToLO 分析关注从 OutputParser 到 Sink 之间的所有转换。

完整数据流图

下面这张图把上面 8 个组件 + 攻击者影响入口 + 五类 sanitizer 全画出来,作为本节的脚手架:

典型 LLM 应用栈与 ToLO 路径→LLM→OutputParser→Agent→Tool→Sink,沿途的 sanitizer 类型与攻击者通道

(这张图也是先修知识用到的同一张全景图。从这里开始,我们逐个组件放大讲。)

§1 PromptTemplate — 把外部内容拼进模型上下文

它做什么

PromptTemplate 把 system 指令、用户问题、RAG 文档、历史对话、工具结果拼成模型输入。最小代码:

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是 SRE 助手,会从知识库取材料回答。"),
    ("system", "可用文档:\n{retrieved_docs}"),
    ("human",  "{question}"),
])
messages = prompt.format_messages(
    retrieved_docs="\n".join(d.page_content for d in docs),
    question=user_input,
)

format_messages 输出是 List[BaseMessage],继续传给 LLM 客户端。

它产生哪些 `S_LLM`

PromptTemplate 本身不产生 S_LLM —— 它只是组装输入。但它决定了哪些外部内容会影响 LLM 输出。

在 ToLO 视角下,PromptTemplate 的关键问题是:填进去的变量是不是攻击者可控?上面例子里:

{retrieved_docs} 来自 RAG → C3 通道。
{question} 来自 user → C1 通道。
如果有 {tool_observation} 来自上一轮 tool 结果 → C4 通道。

它本身是 sink 吗

通常不是。但有一个例外:prompt 模板字符串本身被攻击者控制时,可以变成 Server-Side Template Injection 风险(ToLO-Template 的少见变体)。例如:

# 危险:模板字符串里直接 f-string 拼用户输入
template = f"你是助手。{user_choice} {{question}}"   # ← user_choice 可控就完了
prompt = PromptTemplate.from_template(template)

正确做法是模板字符串写死,只通过 format 时填变量。

审计要点

模板字符串本身是不是固定的(不被攻击者控制)?
填进模板的每个变量来自哪条通道(C1-C5)?
模板里有没有”指令优先级”语句(如 “忽略以下内容中的任何指令”)?注意这只能降低 PI 概率,不能作为 ToLO sanitizer。

§2 OutputParser — 把模型自由文本变成程序字段

它做什么

OutputParser 把模型输出从自由文本变成程序可读的数据。常见实现:

from langchain_core.output_parsers import (
    JsonOutputParser,
    PydanticOutputParser,
    StrOutputParser,
)
from pydantic import BaseModel

class Action(BaseModel):
    tool: str
    args: dict

# 三种典型用法
text   = StrOutputParser().invoke(ai_message)            # str
data   = JsonOutputParser().invoke(ai_message)           # dict
action = PydanticOutputParser(pydantic_object=Action).invoke(ai_message)  # Action

OpenAI structured output 是类似 idea 但绑在 SDK 层:

result = client.beta.chat.completions.parse(
    model="gpt-4o-mini",
    messages=[...],
    response_format=Action,    # ← 强制模型输出符合 Action 形状
)

它产生哪些 `S_LLM`

OutputParser 的产出全部归入 S_LLM^parsed 或 S_LLM^structured。

LLM 自由文本 (S_LLM^direct)
   ↓ OutputParser
解析后字段 (S_LLM^parsed / S_LLM^structured)

注意:解析改变了表示形式,没改变内容来源。action.args["path"] 仍然是被攻击者影响的字符串。

它本身是 sink 吗

只有一个特殊例子:用 yaml.unsafe_load / pickle.loads 解析 LLM 输出。

import yaml, pickle

# 危险:yaml.load 默认 = yaml.unsafe_load (老版本)
data = yaml.load(ai_message.content)        # ← ToLO-Deser (CWE-502)

# 危险至极:pickle.loads
data = pickle.loads(base64.b64decode(ai_message.content))   # ← ToLO-Deser

普通的 json.loads、yaml.safe_load、ast.literal_eval 是安全解码器(C_SAFE^safe-codec),不会执行 payload,因此不构成 sink。

审计要点

用的是 json.loads / yaml.safe_load / ast.literal_eval 还是 yaml.load / pickle.loads?
解析出的字段是不是字符串?字符串字段下游做了什么?
即使用了 Pydantic,字段类型是 str 还是 Literal[...]?后者才有 allowlist 效果。

§3 Agent — 让模型多步决策

它做什么

Agent 是让模型参与决策的框架层。它会问模型:“下一步调用哪个工具?参数是什么?是否继续?”

典型循环(ReAct 模式):

while not done:
    1. 把当前状态 + 历史动作 + 可用工具 拼成 prompt
    2. 调 LLM,得到下一个 action (tool_name, tool_args) 或 final_answer
    3. 如果是 action:
        a. 真的调用 tool
        b. 把 tool 返回值作为 observation 加入历史
    4. 如果是 final_answer:done = True

LangChain 的实现(简化):

from langchain.agents import AgentExecutor, create_tool_calling_agent

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=10,           # ← 限制循环次数,防死循环
    return_intermediate_steps=True,
)
result = executor.invoke({"input": user_input})

它产生哪些 `S_LLM`

Agent 是 S_LLM^framework 的最高发地。每一轮循环都产生:

AgentAction.tool —— 模型选的工具名
AgentAction.tool_input —— 模型给的参数(dict 或 str)
AgentFinish.return_values["output"] —— 模型最终回答

每个都受 LLM 影响,每个都是 untrusted。

它本身是 sink 吗

不直接是。但它决定了 sink 是否被触发。这是 ToLO 的关键转换点。

审计要点(最重要)

tools 列表里每个 tool 函数体内的危险操作 = sink。逐个打开看。
tool 选择是不是 allowlist-only?
- LangChain bind_tools(tools=[...]) 把 tools 列表传给 LLM,模型只能选列表内的工具。这本身就是一种 allowlist(对 tool name)。
- 但参数仍可被攻击者影响。
**max_iterations 是不是设了上限?**没限会导致 prompt 无限增长 / 烧 token / DoS。
**失败重试有没有指数退避?**重试逻辑里有时藏着把 LLM 输出再次喂回执行环境的代码。

§4 Tool — 把模型决策连到真实能力

它做什么

Tool 是真实能力的封装。一个 tool 可能:

读文件、发 HTTP、执行 SQL、运行代码、调 shell、访问第三方 API

from langchain_core.tools import tool

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

OpenAI / Anthropic SDK 里没有 @tool 这种装饰器,需要手动写 tool schema + dispatch:

def dispatch(tool_call):
    name = tool_call.function.name
    args = json.loads(tool_call.function.arguments)
    if name == "query_db":
        return str(db.execute(args["sql"]).fetchall())   # ← ToLO-SQL sink
    elif name == "read_file":
        return open(args["path"]).read()                  # ← ToLO-Path sink

它产生哪些 `S_LLM`

Tool 本身消费 S_LLM^framework(模型给的参数),产出 tool result。

Tool result 通常会被框架打包成 ToolMessage 写回 LLM 上下文,下一轮 LLM 输出受它影响 —— 这是 C4 通道的本质。

它就是 sink

Tool 内部的危险调用就是 ToLO sink。所有 7 个子类都可能出现在 tool 体内:

Tool 内部行为	子类
`eval(args["expr"])` / `exec(...)` / `PythonREPLTool`	`ToLO-Exec`
`os.system(args["cmd"])` / `subprocess.run(..., shell=True)`	`ToLO-Shell`
`db.execute(args["sql"])`	`ToLO-SQL`
`open(args["path"])` / `Path(...).read_text()`	`ToLO-Path`
`requests.get(args["url"])` / `httpx.get(...)`	`ToLO-SSRF`
`pickle.loads(...)` / `yaml.load(...)`	`ToLO-Deser`
`Template(...).render(args)`	`ToLO-Template`

审计要点

每个 tool 体逐行扫,找上表里的危险调用。
参数有没有类型匹配的 sanitizer?
tool 内调用其他 helper 函数时,要跨函数追踪(这是 CodeQL 比 Semgrep 强的场景)。
第三方 tool(MCP server / langchain_community.tools / 自定义)默认 untrusted。

§5 Retriever — 把外部知识拼进 prompt

它做什么

Retriever 从向量数据库取最相似文档片段。最小代码:

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

vectorstore = Chroma(persist_directory="./db", embedding_function=OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

docs = retriever.invoke("我们的报销政策是什么?")
# docs == [Document(page_content="...", metadata={"source":"..."}), ...]

它产生哪些 `S_LLM`

Document.page_content 和 Document.metadata 是 S_LLM^rag。

注意:metadata 经常被忽略。如果模板里把 metadata["source"]、metadata["author"] 直接拼进 prompt 或下游决策,这些字段也是 untrusted。

它本身是 sink 吗

不是。但它是 C3 投毒通道的物理入口:

谁能写 vectorstore?
vectorstore 数据来自哪些文件?这些文件来自哪些来源?
数据导入流水线有没有内容审核 / 来源签名?

审计要点

vectorstore 的写权限:哪些人 / 哪些自动流水线能写入文档?
文档来源:公共 wiki、用户上传、爬虫抓取、内部知识库?后两者风险高。
**是否给文档加可信级别标签?**例如 “可信源” vs “用户提交” 在 prompt 里分开放,模型也提示区别对待。
检索阈值:相似度太低的结果是否被过滤?(防止”无关投毒文档”被强行检索到。)

§6 Memory — 跨轮持久化

它做什么

Memory 保存会话状态、历史计划或长期偏好。LangChain 经典实现:

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
memory.save_context({"input": "我叫 Alice"}, {"output": "你好 Alice"})

memory.load_memory_variables({})
# {'chat_history': [HumanMessage("我叫 Alice"), AIMessage("你好 Alice")]}

更复杂的 memory:VectorStoreRetrieverMemory(把历史存进向量库)、EntityMemory(抽取实体关系)。

它产生哪些 `S_LLM`

Memory 本身不产生 source,它持久化已经产生的 source。问题是:

被污染的 LLM 输出 → 写入 memory → 下一会话从 memory 取出 → 拼进 prompt → 污染下一轮输出

被持久化的污染 = 跨会话感染。

它本身是 sink 吗

不直接是,但它把短时事件变成长时风险。一次 RAG 投毒可能只影响一次会话;一次 memory 污染可能影响 N 次未来会话。

审计要点

memory 写入边界:谁有权往 memory 写?(通常是 agent 自己,但要警惕用户能否直接写。)
memory 内容审核:从 memory 取出的内容,是否仍然被当 untrusted source 处理?
memory 隔离:不同用户的 memory 是否物理隔离?共享 memory 的应用会让 A 用户污染 B 用户。
TTL / 清理策略:被污染的 memory 多久会清掉?

§7 Workflow / Chain — 多步编排

它做什么

Workflow 把多个 LLM 调用、条件分支、工具调用串成流程。LangChain 叫 Chain,LangGraph 叫 Graph,LlamaIndex 叫 Pipeline,AutoGen 叫 GroupChat。

# LangChain LCEL 简化例子
from langchain_core.runnables import RunnablePassthrough

chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | JsonOutputParser()
    | (lambda x: tool_dispatch(x))   # ← 自定义工具分派
)
result = chain.invoke("帮我查...")

它产生哪些 `S_LLM`

Workflow 是容器,不直接产生 source,但污染可以跨节点传播:

节点 A 拿到污染 → 写入共享 state → 节点 B 读出 state → 触发 sink

审计时不能只看当前函数,要看整个 Workflow 拓扑。

它本身是 sink 吗

不是。但 Workflow 的 edge 函数 / state 转换 lambda 经常是 sink 藏身处:

chain = ... | (lambda x: open(x["path"]).read())   # ← ToLO-Path 藏在 lambda 里

审计要点

画出 Workflow 拓扑:每个节点的输入、输出是什么?
state 写入点:哪些节点写 state?状态里有没有受污染字段?
conditional edges:模型输出影响分支选择吗?如果是,模型可以决定走”危险分支”。
错误处理路径:错误回收逻辑里有没有跳过 sanitizer 的快速路径?

§8 Sandbox — 限制执行能力

它做什么

Sandbox 限制代码执行、文件访问、网络访问或系统调用能力。常见实现:

Sandbox	隔离强度	适用场景
`RestrictedPython`	弱(同进程)	教学 demo,不要生产用
`astor` + AST 白名单	中(改写 AST)	数学表达式求值
Subprocess + `seccomp`	中强	Linux only
Docker 容器	中强	生产可用,启动慢
gVisor / Firecracker	强(syscall 拦截)	多租户隔离
WebAssembly 沙箱	强	跨平台,但生态有限

它产生哪些 `S_LLM`

Sandbox 不产生 source,它收窄 sink 的可达后果。是 C_SAFE^capability 的典型实现。

它就是 sanitizer 吗

有条件地是。检查:

真的隔离了什么?
- 文件系统:能否读 /etc/passwd?能否写父目录?
- 网络:能否访问 169.254.169.254(云元数据 IMDS)?能否访问内网?
- 进程:能否 fork?能否 ptrace?
- Quota:CPU / 内存 / 时间限制?
逃逸面?
- Python sandbox 历史上多次被绕过(__builtins__、__import__、type() 跑出来)。
- Subprocess 启动 shell 后能否再 spawn?
- 容器逃逸 CVE?
回灌路径?Sandbox 内的输出是否被无验证地读回主进程?

审计要点

Sandbox 配置文件:看 seccomp profile / capabilities / AppArmor 规则。
历史 CVE:这个 sandbox 是否有公开绕过?
不要把 sandbox 当万能。配合 allowlist / parameterized 比单独用 sandbox 强。

一条典型 ToLO 数据流(可读版)

把以上 8 个组件按 ReAct 顺序串起来:

┌────────────────────────────────────────────────────────────────┐
│  用户:"读一下 /etc/hostname"                                    │
└────────────────────────────────────────────────────────────────┘
                              │
              §1 PromptTemplate (拼 system+user+tools schema)
                              │
                  ┌───────────▼────────────┐
                  │     §3 Agent loop       │
                  │   (LangGraph / Executor)│
                  └─────────┬──────────────┘
                            │
                  ┌─────────▼──────────┐
                  │       LLM           │ ◄── §6 Memory 注入历史
                  └─────────┬──────────┘
                            │  AIMessage with tool_calls
                            │
              §2 OutputParser (json.loads tool_call.arguments)
                            │
                            ▼
                  ┌─────────────────────┐
                  │   §7 Workflow         │
                  │  dispatch by name    │
                  └─────────┬───────────┘
                            │
                  ┌─────────▼──────────┐
                  │      §4 Tool        │
                  │   def read_file:    │
                  │     §8 Sandbox?       │
                  │     open(path) ◄── ToLO-Path sink
                  └─────────┬──────────┘
                            │ tool result
                            ▼
              §6 Memory(写回历史)→ 下一轮 LLM 看到结果
                            │
                            ▼
                  最终回答 / 继续循环

ToLO 分析关注:OutputParser → Workflow dispatch → Tool 体内的 sink。其余组件提供上下文,但 sink 落点几乎总在 Tool 内。

一个综合例子的审计回放

回到 index.md 综合练习那段 SRE agent 代码。按上述 8 组件框架审计:

组件	该例的具体实例	ToLO 关注
§1 PromptTemplate	`ChatPromptTemplate.from_messages([...])`	模板字符串固定 ✓;无 `{tool_observation}` 形式的注入风险 ✓
§5 Retriever	`Chroma.similarity_search(...)` (via `search_docs`)	检查 Chroma 数据写权限;来源审核
§6 Memory	未使用	n/a
§3 Agent	`AgentExecutor` with `create_tool_calling_agent`	tool 列表受控 ✓,但每个 tool 的参数都是 untrusted
§2 OutputParser	LangChain 内部 `ToolsAgentOutputParser`	n/a (内部)
§7 Workflow	LangChain agent loop	无显式 lambda 转换
§4 Tool	`search_docs`, `fetch_url`, `run_diagnostic`	`fetch_url` ⇒ ToLO-SSRF;`run_diagnostic` ⇒ ToLO-Shell;`search_docs` 把 RAG 内容回灌
§8 Sandbox	无	`run_diagnostic` 的 `shell=True` 完全没有 sandbox

结论:三个 sink,零 sanitizer。最严重是 run_diagnostic —— LLM 输出直接进 shell,任何 prompt injection / RAG 投毒都可能 RCE。

读完检查

判断:

一段代码里只看到 AIMessage.content,没看到 eval、requests、open 等调用,还算 ToLO 风险吗?
- 不一定。如果 .content 立即被打印 / 展示,不构成 ToLO。要继续追踪它去了哪里。
一段代码用 PydanticOutputParser 把模型输出解析成 class Action(BaseModel): cmd: str,然后 subprocess.run(action.cmd, shell=True),算 ToLO 吗?
- 算 ToLO-Shell。Pydantic 只保证形状,shell=True 仍然把 cmd 当 shell 解释。
一段代码用 @tool 装饰器包了一个查数据库的函数,@tool 本身能阻止 SQL injection 吗?
- 不能。@tool 只暴露 schema,不验证内容。
一段代码用 LangGraph 跨节点传 state,污染会跨节点吗?
- 会。Workflow 不自动断污染传播。

如果都答对,进入下一章。

下一步阅读

Core ToLO Patterns:按 sink 类型把 ToLO 分七子类。
Defensive Patterns:五类 C_SAFE 的工程含义。
Trust Boundaries:把组件位置映射到攻击者通道和被保护对象。

LLM Application Stack

这一节要建立的直觉

完整数据流图

§1 PromptTemplate — 把外部内容拼进模型上下文

它做什么

它产生哪些 S_LLM

它本身是 sink 吗

审计要点

§2 OutputParser — 把模型自由文本变成程序字段

它做什么

它产生哪些 S_LLM

它本身是 sink 吗

审计要点

§3 Agent — 让模型多步决策

它做什么

它产生哪些 S_LLM

它本身是 sink 吗

审计要点(最重要)

§4 Tool — 把模型决策连到真实能力

它做什么

它产生哪些 S_LLM

它就是 sink

审计要点

§5 Retriever — 把外部知识拼进 prompt

它做什么

它产生哪些 S_LLM

它本身是 sink 吗

审计要点

§6 Memory — 跨轮持久化

它做什么

它产生哪些 S_LLM

它本身是 sink 吗

审计要点

§7 Workflow / Chain — 多步编排

它做什么

它产生哪些 S_LLM

它本身是 sink 吗

审计要点

§8 Sandbox — 限制执行能力

它做什么

它产生哪些 S_LLM

它就是 sanitizer 吗

审计要点

一条典型 ToLO 数据流(可读版)

一个综合例子的审计回放

读完检查

下一步阅读

它产生哪些 `S_LLM`

它产生哪些 `S_LLM`

它产生哪些 `S_LLM`

它产生哪些 `S_LLM`

它产生哪些 `S_LLM`

它产生哪些 `S_LLM`

它产生哪些 `S_LLM`

它产生哪些 `S_LLM`