1
while not done:
2
    thought = llm.reason(context)
3
    action = llm.choose_tool(thought)
4
    observation = execute(action)
5
    context.append(observation)

1
# 输入任务
2
task = "研究 Rust 的异步编程模型并撰写技术博客"
3

4
# LLM 生成的计划
5
plan = [
6
    {
7
        "step": 1,
8
        "action": "搜索 Rust async/await 官方文档",
9
        "expected_output": "核心概念清单"
10
    },
11
    {
12
        "step": 2,
13
        "action": "搜索 Tokio 运行时实现细节",
14
        "expected_output": "技术架构图"
15
    },
16
    {
17
        "step": 3,
18
        "action": "对比 Go/Node.js 的异步模型",
19
        "expected_output": "差异分析表格"
20
    },
21
    {
22
        "step": 4,
23
        "action": "撰写博客大纲",
24
        "expected_output": "Markdown 文件"
25
    }
26
]

1
主 Agent（Coordinator）
2
    ├─ 负责任务分解和结果汇总
3
    └─ 不执行具体工作
4

5
子 Agent（Workers）
6
    ├─ 专注于单一子任务
7
    ├─ 独立的上下文窗口
8
    └─ 可以并行运行

1
class CoordinatorAgent:
2
    def execute(self, task):
3
        # 1. 规划
4
        plan = self.planning_tool.decompose(task)
5

6
        # 2. 并行委派
7
        futures = []
8
        for subtask in plan:
9
            worker = self.spawn_worker(subtask.type)
10
            future = worker.execute_async(subtask)
11
            futures.append(future)
12

13
        # 3. 汇总结果
14
        results = await asyncio.gather(*futures)
15
        return self.synthesize(results)

1
graph TB
2
    subgraph Episodic[情景记忆 Episodic Memory]
3
        E1[对话历史]
4
        E2[操作日志]
5
        E3[执行轨迹]
6
    end
7

8
    subgraph Semantic[语义记忆 Semantic Memory]
9
        S1[知识图谱]
10
        S2[概念索引]
11
        S3[向量数据库]
12
    end
13

14
    subgraph Procedural[程序记忆 Procedural Memory]
15
        P1[工作流模板]
16
        P2[最佳实践]
17
        P3[策略库]
18
    end
19

20
    Agent[Deep Agent] --> Episodic
21
    Agent --> Semantic
22
    Agent --> Procedural
23

24
    Episodic --> ShortTerm[短期存储<br/>会话级别]
25
    Semantic --> LongTerm[长期存储<br/>持久化]
26
    Procedural --> LongTerm
27

28
    style Agent fill:#C4612F,color:#fff
29
    style Episodic fill:#F2E3D6
30
    style Semantic fill:#F2E3D6
31
    style Procedural fill:#F2E3D6

1
Thought → Action → Observation → (循环)

1
graph LR
2
    Plan[规划 Plan] --> Reason[推理 Reason]
3
    Reason --> Act[行动 Act]
4
    Act --> Observe[观察 Observe]
5
    Observe --> Reflect[反思 Reflect]
6
    Reflect --> UpdatePlan[更新计划]
7
    UpdatePlan --> Reason
8

9
    Reflect -.-> Memory[(记忆系统)]
10
    Memory -.-> Reason
11

12
    style Plan fill:#C4612F,color:#fff
13
    style Reflect fill:#F2E3D6
14
    style Memory fill:#F2E3D6

1
Agent (Thought): 我需要搜索 Rust async 的实现原理
2
Agent (Action): web_search("Rust async implementation")
3
Agent (Observation): 返回了 5 篇文章，但都是入门教程
4
Agent (Reflection): 这些结果太浅显，我需要更深入的技术文档
5
Agent (Update Plan): 修改搜索关键词为 "Rust Future trait internals"

1
def execute_with_reflexion(task):
2
    # 1. 检索历史反思
3
    past_failures = memory.query_similar(task)
4

5
    # 2. 增强提示词
6
    prompt = f"""
7
    Task: {task}
8

9
    Previous Failures:
10
    {past_failures}
11

12
    Learn from these mistakes and avoid them.
13
    """
14

15
    # 3. 执行
16
    result = agent.run(prompt)
17

18
    # 4. 如果失败，生成反思
19
    if result.failed:
20
        reflection = llm.reflect(task, result.error)
21
        memory.store(reflection)
22

23
    return result

1
from langgraph.graph import StateGraph, END
2
from langchain_anthropic import ChatAnthropic
3
from langchain.tools import DuckDuckGoSearchRun, WikipediaQueryRun
4

5
# 定义状态
6
class ResearchState:
7
    task: str
8
    plan: List[dict]
9
    current_step: int
10
    memory: List[str]
11
    artifacts: dict  # 文件系统
12

13
# 1. 规划节点
14
def planning_node(state: ResearchState):
15
    llm = ChatAnthropic(model="claude-opus-4")
16
    prompt = f"分解任务为步骤：{state.task}"
17
    plan = llm.invoke(prompt)  # 返回结构化 JSON
18
    state.plan = plan
19
    return state
20

21
# 2. 执行节点（并行 Worker）
22
async def execution_node(state: ResearchState):
23
    step = state.plan[state.current_step]
24

25
    # 根据步骤类型选择 Worker
26
    if step["type"] == "web_search":
27
        tool = DuckDuckGoSearchRun()
28
        result = await tool.arun(step["query"])
29
    elif step["type"] == "wiki_query":
30
        tool = WikipediaQueryRun()
31
        result = await tool.arun(step["query"])
32

33
    # 存储到文件系统
34
    state.artifacts[f"step_{state.current_step}.md"] = result
35
    state.current_step += 1
36
    return state
37

38
# 3. 反思节点
39
def reflection_node(state: ResearchState):
40
    # 检查当前结果是否满足预期
41
    last_result = state.artifacts[f"step_{state.current_step-1}.md"]
42
    llm = ChatAnthropic(model="claude-opus-4")
43

44
    reflection = llm.invoke(f"""
45
    任务：{state.task}
46
    当前结果：{last_result}
47

48
    评估：结果是否充分？是否需要调整计划？
49
    """)
50

51
    if "需要重新搜索" in reflection:
52
        # 动态调整计划
53
        state.plan.insert(state.current_step, {
54
            "type": "web_search",
55
            "query": reflection["建议关键词"]
56
        })
57

58
    return state
59

60
# 4. 构建图
61
workflow = StateGraph(ResearchState)
62
workflow.add_node("plan", planning_node)
63
workflow.add_node("execute", execution_node)
64
workflow.add_node("reflect", reflection_node)
65

66
workflow.add_edge("plan", "execute")
67
workflow.add_edge("execute", "reflect")
68
workflow.add_conditional_edge(
69
    "reflect",
70
    lambda state: "execute" if state.current_step < len(state.plan) else END
71
)
72

73
workflow.set_entry_point("plan")
74
agent = workflow.compile()
75

76
# 5. 运行
77
result = await agent.ainvoke({
78
    "task": "研究 Rust 的 async/await 实现原理",
79
    "plan": [],
80
    "current_step": 0,
81
    "memory": [],
82
    "artifacts": {}
83
})

1
graph LR
2
    Request[请求] --> Auth[权限中间件]
3
    Auth --> Logger[日志中间件]
4
    Logger --> Memory[记忆中间件]
5
    Memory --> Core[核心 Agent]
6
    Core --> Validation[验证中间件]
7
    Validation --> Response[响应]
8

9
    Auth -.-> Reject1[拒绝]
10
    Validation -.-> Reject2[重试]
11

12
    style Core fill:#C4612F,color:#fff
13
    style Auth fill:#F2E3D6
14
    style Memory fill:#F2E3D6
15
    style Validation fill:#F2E3D6