塞满即崩：六栈上下文工程实测

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph.message import add_messages
from typing import Annotated, TypedDict
from langchain_core.messages import SystemMessage, trim_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]
    summary: str                       # 压缩产物独立存放

def call_model(state: State):
    # compress 策略：每次只喂精挑后的窗口
    trimmed = trim_messages(
        state["messages"], strategy="last",
        max_tokens=4000, include_system=True, start_on="human",
    )
    msgs = [SystemMessage(state["summary"])] + trimmed if state["summary"] else trimmed
    return {"messages": [llm.invoke(msgs)]}

def summarize(state: State):
    s = llm.invoke([SystemMessage("摘要以下历史:")] + state["messages"][:-2])
    return {"summary": s.content, "messages": state["messages"][-2:]}

builder = StateGraph(State)
builder.add_node("model", call_model)
builder.add_node("summarize", summarize)
builder.add_conditional_edges("model",
    lambda s: "summarize" if len(s["messages"]) > 10 else END)
builder.add_edge("summarize", "model")
graph = builder.compile(checkpointer=MemorySaver())

from agents import Agent, Runner, RunContextWrapper
from dataclasses import dataclass

@dataclass
class UserCtx:           # 本地上下文：依赖注入，LLM 看不到
    user_id: str
    db_pool: object

async def fetch_orders(ctx: RunContextWrapper[UserCtx], query: str):
    # 工具内可直接拿到 db_pool，无需塞进 prompt
    return await ctx.context.db_pool.fetch(query)

triage = Agent(name="分诊", tools=[fetch_orders],
               handoffs=[billing_agent])
result = await Runner.run(trage, "查我上个月账单", context=UserCtx("u1", pool))

from crewai import Agent, Task, Crew

researcher = Agent(role="研究员",
    backstory="资深行业分析师，只信一手数据",
    goal="产出 3 条带出处的洞察")

writer = Agent(role="撰稿人",
    backstory="把数据翻译成高管能 30 秒读完的摘要",
    goal="输出 200 字结论")

t1 = Task(description="调研{topic}最新进展", agent=researcher,
          expected_output="3 条洞察")
t2 = Task(description="将洞察改写为摘要", agent=writer,
          expected_output="200 字", context=[t1])   # select：只喂 t1 的产物

crew = Crew(agents=[researcher, writer], tasks=[t1, t2])

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_core.models import ChatCompletionClient

# 压缩器：对话超过阈值自动 condense
from autogen_agentchat.condensers import LLMSummaryCondenser

model = ChatCompletionClient(model="gpt-4o")
condenser = LLMSummaryCondenser(
    model_client=model, max_messages=20,   # 超过 20 条触发摘要
    trigger_messages=15)

agent = AssistantAgent("analyst", model_client=model,
    system_message="你是严谨的分析师",)
team = RoundRobinGroupChat([agent],
    termination_condition=TextMentionTermination("TERMINATE"))

import anyio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Grep", "Task"],   # select：限定可见工具
        # compaction 由 SDK 在阈值自动触发，无需手写
        mcp_servers={                              # 外置 context（MCP）
            "db": {"command": "uvx", "args": ["db-mcp"]}},
        setting_sources=["project", "local"],      # 写：CLAUDE.md 持久指令
    )
    async for msg in query(prompt="重构 src/auth", options=options):
        print(msg)

anyio.run(main)

from letta_client import Letta

client = Letta(base_url="http://localhost:8283")
agent = client.agents.create(
    name="长记忆伴侣",
    memory_blocks=[
        {"label": "human", "value": "用户偏好：简洁、爱用数据说话"},
        {"label": "persona", "value": "我是严谨的助理"},
    ],
    model="anthropic/claude-sonnet-4-6",
    # archival memory 自动分页，超长对话不爆窗口
)

# agent 运行中可自主调用 archival_memory_search 取回旧信息
response = client.agents.messages.create(
    agent_id=agent.id,
    messages=[{"role": "user", "content": "我们三个月前聊过的那个方案"}])

flowchart TB
    subgraph 四策略["上下文工程四策略 (Anthropic)"]
        W["✍️ Write 写入<br/>该进窗口的持久化"]
        S["🔍 Select 选取<br/>每步按需召回"]
        C["🗜️ Compress 压缩<br/>摘要/裁剪历史"]
        I["🧱 Isolate 隔离<br/>子上下文分治"]
    end
    subgraph 六栈["六栈实现手法"]
        LG["LangGraph<br/>State+检查点+trim"]
        OA["OpenAI SDK<br/>DI+Handoff"]
        CR["CrewAI<br/>角色作用域"]
        AT["AutoGen<br/>Condenser摘要"]
        CA["Claude SDK<br/>compaction+子上下文"]
        LE["Letta<br/>虚拟内存分页"]
    end
    W --> LG & CA & LE
    S --> OA & CR & LE
    C --> LG & AT & CA
    I --> LG & OA & CR & CA
    classDef strat fill:#e8f4fd,stroke:#2563eb,stroke-width:2px
    classDef stack fill:#fef3c7,stroke:#d97706,stroke-width:1px
    class W,S,C,I strat
    class LG,OA,CR,AT,CA,LE stack

sequenceDiagram
    participant U as 用户
    participant Orch as 编排Agent
    participant Sub as 子Agent
    participant Mem as 外部记忆
    U->>Orch: 长任务请求
    Note over Orch: ✍️ Write: CLAUDE.md/State 持久指令
    Orch->>Mem: 🔍 Select: 按需召回相关片段
    Mem-->>Orch: 注入精选上下文
    Orch->>Sub: 🧱 Isolate: 派发+裁剪上下文
    Note over Sub: 子上下文独立工作
    Sub-->>Orch: 只回传摘要结果
    Note over Orch: 🗜️ Compress: 超阈值自动摘要旧轮次
    Orch-->>U: 持续输出，窗口不爆