nanobot 源码解读 · 03 | 上下文工程：一条消息如何被拼成 LLM 能读懂的「剧本」

flowchart TD
    A["通道来的一条用户消息<br/>+ media + runtime_context_blocks"] --> B["AgentLoop._build_initial_messages<br/>(loop.py:711)"]
    B --> C["ContextBuilder.build_messages<br/>(context.py:206)"]
    C --> D["① 拼 system prompt<br/>build_system_prompt (context.py:70)"]
    C --> E["② 拼 history<br/>(会话历史，来自 SessionStore)"]
    C --> F["③ 拼当前轮<br/>build_current_message (context.py:265)"]
    D & E & F --> G["initial_messages<br/>(SDK 不存，只作这次请求的起点)"]
    G --> H["AgentRunner._run_core 循环<br/>(runner.py:419)"]
    H --> I["每一轮: ContextGovernor.prepare_for_model<br/>(context_governance.py:76)"]
    I --> J["修干净+压小后的 messages_for_model"]
    J --> K["_request_model 调 LLM"]
    K --> L["工具结果 append 回 messages (持久化副本)"]
    L -.下一轮.-> I

def build_messages(
    self,
    history: list[dict[str, Any]],
    current_message: str,
    *,
    media: list[str] | None = None,
    channel: str | None = None,
    current_role: str = "user",
    session_summary: str | None = None,
    runtime_context_blocks: Sequence[RuntimeContextBlock] | None = None,
    workspace: Path | None = None,
    include_memory_recent_history: bool = True,
    session_key: str | None = None,
    unified_session: bool = False,
) -> list[dict[str, Any]]:
    root = workspace or self.workspace
    # 1) 从用户当前文本里，挑出被"显式 @调用"的 skill 名字
    active_skill_names = (
        self.skills.get_explicitly_invoked_skills(current_message)
        if current_role == "user"
        else []
    )
    # 2) 第一条永远是 system，里面是所有"背景知识"
    messages: list[dict[str, Any]] = [
        {
            "role": "system",
            "content": self.build_system_prompt(
                active_skill_names=active_skill_names,
                channel=channel,
                session_summary=session_summary,
                workspace=root,
                include_memory_recent_history=include_memory_recent_history,
                session_key=session_key,
                unified_session=unified_session,
            ),
        },
        *history,   # 3) 然后是持久化会话历史（user/assistant/tool 交替）
    ]
    # 4) 当前这轮的 user 消息
    current = self.build_current_message(
        current_message,
        media=media,
        current_role=current_role,
        runtime_context_blocks=runtime_context_blocks,
    )
    # 5) 如果 history 最后一条恰好也是 user（比如上一轮被打断），合并而不是再加一条
    if messages[-1].get("role") == current_role:
        last = dict(messages[-1])
        last["content"] = self._merge_message_content(
            last.get("content"), current.get("content"),
        )
        messages[-1] = last
        return messages
    messages.append(current)
    return messages

def build_system_prompt(self, *, active_skill_names=None, channel=None,
                        session_summary=None, workspace=None,
                        include_memory_recent_history=True,
                        session_key=None, unified_session=False) -> str:
    root = workspace or self.workspace
    parts = [self._get_identity(channel=channel, workspace=root)]   # ① 身份

    bootstrap = self._load_bootstrap_files(root)                    # ② AGENTS/SOUL/USER.md
    if bootstrap:
        parts.append(bootstrap)

    parts.append(render_template("agent/tool_contract.md"))         # ③ 工具契约说明

    memory = self.memory.read_memory()                              # ④ 长期记忆 MEMORY.md
    if memory and not self._is_template_content(memory, "memory/MEMORY.md"):
        parts.append(f"# Memory\n\n## Long-term Memory\n{memory}")

    # ⑤ 本次被显式 @的 skill 全文 + ⑥ 其余 skill 的摘要清单
    active_skills = self.skills.get_always_skills()
    active_skills.extend(name for name in (active_skill_names or ()) if name not in active_skills)
    if active_skills:
        active_content = self.skills.load_skills_for_context(active_skills)
        if active_content:
            parts.append(f"# Active Skills\n\n{active_content}")
    skills_summary = self.skills.build_skills_summary(exclude=set(active_skills))
    if skills_summary:
        parts.append(render_template("agent/skills_section.md", skills_summary=skills_summary))

    # ⑦ 长期记忆里"最近发生的事"（Dream 之后还没沉淀的），带 token 上限
    if include_memory_recent_history:
        entries = self.memory.read_recent_history_for_prompt(
            since_cursor=self.memory.get_last_dream_cursor(),
            session_key=session_key, unified_session=unified_session,
        )
        if entries:
            capped = entries[-self._MAX_RECENT_HISTORY:]             # 最近 50 条
            history_text = "\n".join(f"- [{e['timestamp']}] {e['content']}" for e in capped)
            history_text = truncate_text_to_tokens(history_text, self._MAX_HISTORY_TOKENS)  # 上限 8000 token
            parts.append("# Recent History\n\n" + history_text)

    # ⑧ 会话压缩摘要（被 loop 提前算好，作为 [Archived Context Summary] 塞进来）
    if session_summary:
        parts.append(f"[Archived Context Summary]\n\n{session_summary}")

    return "\n\n---\n\n".join(parts)

workspace_path = str(root.expanduser().resolve())
agent_workspace_path = str(self.workspace.expanduser().resolve())
system = platform.system()
runtime = f"{'macOS' if system == 'Darwin' else system} {platform.machine()}, Python {platform.python_version()}"

for path in image_paths:
    p = Path(path)
    if not p.is_file():
        continue
    raw = p.read_bytes()
    mime = detect_image_mime(raw) or mimetypes.guess_type(path)[0]   # 重新探测 MIME
    if not mime or not mime.startswith("image/"):
        continue
    b64 = base64.b64encode(raw).decode()
    image_blocks.append({
        "type": "image_url",
        "image_url": {"url": f"data:{mime};base64,{b64}"},
        "_meta": {"path": str(p)},
    })

def prepare_for_model(self, config, messages, compacted_tool_call_ids):
    updated = self.strip_placeholder_assistant_messages(messages)  # 1) 剥占位符
    updated = self.strip_malformed_tool_calls(updated)             # 2) 剔畸形 tool_call
    updated = self.drop_orphan_tool_results(updated)               # 3) 丢孤儿 tool 结果
    updated = self.backfill_missing_tool_results(updated)          # 4) 补缺失结果
    updated = self.apply_tool_result_budget(config, updated)       # 5) 工具结果预算/落盘
    updated = self.compact_inflight_overflow(config, updated, compacted_tool_call_ids)  # 6) 在飞压缩
    updated = self.snip_history(config, updated)                   # 7) 历史裁剪
    updated = self.drop_orphan_tool_results(updated)               # 8) 再丢一次孤儿
    return self.backfill_missing_tool_results(updated)             # 9) 再补一次

for iteration in range(spec.max_iterations):
    # 持久化历史不被治理改动；治理只产出"模型副本"
    messages_for_model = self.context_governor.prepare_for_model(
        governance_config, messages, compacted_tool_call_ids,
    )
    ...
    response = await self._request_model(spec, messages_for_model, ...)

def normalize_tool_result(self, config, tool_call_id, tool_name, result):
    result = ensure_nonempty_tool_result(tool_name, result)   # 空结果补成提示
    if tool_name in TOOL_RESULT_OFFLOAD_EXEMPT_TOOLS:          # read_file 豁免
        return result
    try:
        content = maybe_persist_tool_result(                   # 超长结果落盘成文件
            config.workspace, config.session_key, tool_call_id, result,
            max_chars=config.max_tool_result_chars,
        )
    except Exception:
        content = result
    if isinstance(content, str) and len(content) > config.max_tool_result_chars:
        return truncate_text(content, config.max_tool_result_chars)
    return content

budget = self.input_budget(config)
if budget <= 0:
    return messages
tools = config.tools.get_definitions()
updated = self._apply_recorded_compactions(messages, compacted_tool_call_ids)  # 先应用上次已压的
estimate, source = estimate_prompt_tokens_chain(config.provider, config.model, updated, tools)
if estimate <= budget:
    return updated              # 没超，原样返回，零开销

target = int(budget * INFLIGHT_COMPACT_TARGET_RATIO)   # 0.85，压到预算的 85%
candidates = self._inflight_compaction_candidates(config, updated, compacted_tool_call_ids)
for candidate_idx, (idx, tool_call_id) in enumerate(candidates):
    ...
    compacted_tool_call_ids.add(tool_call_id)
    self._compact_tool_result_at(updated, idx)         # 把工具结果替换成一句提示
    estimate, source = estimate_prompt_tokens_chain(...)
    if estimate <= target:
        break

COMPACTABLE_TOOLS = frozenset({
    "read_file", "exec", "grep", "find_files",
    "web_search", "web_fetch", "list_dir", "list_exec_sessions",
})
MICROCOMPACT_MIN_CHARS = 500
...
# 只压 inflight_start_index 之后（即本次 turn 产生的）、名字在白名单、且内容 > 500 字符的工具结果
if idx < config.inflight_start_index:
    continue
if msg.get("role") != "tool" or msg.get("name") not in COMPACTABLE_TOOLS:
    continue
if not tool_call_id or str(tool_call_id) in compacted_tool_call_ids:
    continue
content = msg.get("content")
if not isinstance(content, str) or len(content) < MICROCOMPACT_MIN_CHARS:
    continue

Error: The previous exec result was compacted to fit context because it was too large.
Do not repeat the same call unchanged. Retry with a narrower path, query, range, or
result limit, use another tool, or tell the user the task cannot fit in the available context.

system_messages = [dict(msg) for msg in messages if msg.get("role") == "system"]
non_system = [dict(msg) for msg in messages if msg.get("role") != "system"]
system_tokens = sum(estimate_message_tokens(msg) for msg in system_messages)
remaining_budget = max(0, budget - max(system_tokens, fixed_tokens))
kept: list[dict] = []
kept_tokens = 0
for message in reversed(non_system):            # 从最新往最旧扫
    msg_tokens = estimate_message_tokens(message)
    if kept and kept_tokens + msg_tokens > remaining_budget:
        break
    kept.append(message)
    kept_tokens += msg_tokens
kept.reverse()
return system_messages + self._legal_history_tail(kept, non_system)

initial_messages = [
  {role: system, content: "你是 nanobot...（身份）
    ---
    ## AGENTS.md ...（项目指令）
    ---
    ## 工具契约 ...
    ---
    # Memory ...（长期记忆）
    ---
    # Active Skills ...
    ---
    # Recent History ...（最近 50 条）
    ---
    [Archived Context Summary] ...（若有压缩摘要）"},
  {role: user, content: [ {image_url: 报错截图}, {text: "帮我读..."} ]},  # 当前轮（图片已转 base64）
]