1
┌─────────────────────────────────────────────────────────────┐
2
│                    传统 Agent 的困境                           │
3
├─────────────────────────────────────────────────────────────┤
4
│                                                             │
5
│  1. 记忆缺失                                                 │
6
│     ├── 无法记住过去的经历                                    │
7
│     ├── 每次交互都是「白纸一张」                              │
8
│     └── 缺乏连贯性和个性化                                    │
9
│                                                             │
10
│  2. 社交缺失                                                 │
11
│     ├── 大多数研究聚焦单 Agent                               │
12
│     ├── 忽视 Agent 间的互动                                  │
13
│     └── 无法模拟真实社会行为                                  │
14
│                                                             │
15
└─────────────────────────────────────────────────────────────┘

1
感知（Perceive）→ 记忆（Retrieve）→ 反思（Reflect）→ 规划（Plan）→ 行动（Act）

1
┌─────────────────────────────────────────────────────────────┐
2
│                    Memory Stream 结构                         │
3
├─────────────────────────────────────────────────────────────┤
4
│                                                             │
5
│  每条记忆包含：                                               │
6
│  ┌─────────────────────────────────────────────────────┐   │
7
│  │  内容（Content）："John 在咖啡厅遇到了 Maria"          │   │
8
│  │  时间戳（Timestamp）：2024-03-15 14:30:00            │   │
9
│  │  重要性（Importance）：0.8                           │   │
10
│  │  访问次数（Access Count）：3                         │   │
11
│  │  创建时间（Created）：2024-03-15 14:30:05            │   │
12
│  └─────────────────────────────────────────────────────┘   │
13
│                                                             │
14
│  记忆按时间顺序存储，支持智能检索                              │
15
│                                                             │
16
└─────────────────────────────────────────────────────────────┘

1
from dataclasses import dataclass
2
from datetime import datetime
3
from typing import List, Optional
4
import json
5

6
@dataclass
7
class Memory:
8
    """单条记忆"""
9
    content: str              # 记忆内容
10
    timestamp: datetime       # 事件发生时间
11
    created_at: datetime      # 记忆创建时间
12
    importance: float         # 重要性评分 (0-1)
13
    access_count: int = 0     # 访问次数
14
    embedding: List[float] = None  # 向量嵌入（用于检索）
15

16
    def to_dict(self) -> dict:
17
        return {
18
            "content": self.content,
19
            "timestamp": self.timestamp.isoformat(),
20
            "created_at": self.created_at.isoformat(),
21
            "importance": self.importance,
22
            "access_count": self.access_count
23
        }
24

25

26
class MemoryStream:
27
    """记忆流：Agent 的长期记忆系统"""
28

29
    def __init__(self, embedding_model=None):
30
        self.memories: List[Memory] = []
31
        self.embedding_model = embedding_model
32

33
    def add_memory(self, content: str, importance: float = 0.5):
34
        """添加新记忆"""
35
        now = datetime.now()
36
        memory = Memory(
37
            content=content,
38
            timestamp=now,
39
            created_at=now,
40
            importance=importance,
41
            embedding=self._get_embedding(content) if self.embedding_model else None
42
        )
43
        self.memories.append(memory)
44
        return memory
45

46
    def _get_embedding(self, text: str) -> List[float]:
47
        """获取文本嵌入向量"""
48
        if self.embedding_model:
49
            return self.embedding_model.embed(text)
50
        return []

1
import numpy as np
2
from datetime import datetime, timedelta
3

4
class MemoryStreamWithRetrieval(MemoryStream):
5
    """带智能检索的记忆流"""
6

7
    def retrieve(self, query: str, top_k: int = 10) -> List[Memory]:
8
        """检索相关记忆"""
9
        query_embedding = self._get_embedding(query)
10

11
        scored_memories = []
12
        for memory in self.memories:
13
            # 计算三维评分
14
            relevance = self._calculate_relevance(query_embedding, memory)
15
            recency = self._calculate_recency(memory)
16
            importance = memory.importance
17

18
            # 综合评分（论文中的权重）
19
            score = (
20
                0.5 * relevance +
21
                0.3 * recency +
22
                0.2 * importance
23
            )
24
            scored_memories.append((memory, score))
25

26
        # 排序并返回 Top-K
27
        scored_memories.sort(key=lambda x: x[1], reverse=True)
28

29
        # 更新访问计数
30
        for memory, _ in scored_memories[:top_k]:
31
            memory.access_count += 1
32

33
        return [m for m, s in scored_memories[:top_k]]
34

35
    def _calculate_relevance(self, query_embedding: List[float],
36
                            memory: Memory) -> float:
37
        """计算相关性：查询与记忆的语义相似度"""
38
        if memory.embedding is None or query_embedding is None:
39
            return 0.5
40

41
        # 余弦相似度
42
        dot_product = np.dot(query_embedding, memory.embedding)
43
        norm = np.linalg.norm(query_embedding) * np.linalg.norm(memory.embedding)
44
        return dot_product / norm if norm > 0 else 0
45

46
    def _calculate_recency(self, memory: Memory) -> float:
47
        """计算时近性：记忆越新，分数越高"""
48
        hours_ago = (datetime.now() - memory.timestamp).total_seconds() / 3600
49
        # 指数衰减
50
        decay_factor = 0.99 ** hours_ago
51
        return decay_factor
52

53
    def _calculate_importance(self, memory: Memory) -> float:
54
        """重要性已在存储时计算"""
55
        return memory.importance

1
def assess_importance(llm, event: str) -> float:
2
    """使用 LLM 评估事件重要性"""
3
    prompt = f"""
4
    On the scale of 1 to 10, where 1 is purely mundane
5
    (e.g., brushing teeth, making bed) and 10 is
6
    extremely poignant (e.g., a break up, college
7
    acceptance), rate the likely poignancy of the
8
    following piece of memory.
9

10
    Memory: {event}
11

12
    Rating (1-10):
13
    """
14

15
    response = llm.generate(prompt)
16
    # 提取数字并归一化到 0-1
17
    try:
18
        rating = float(response.strip())
19
        return min(max(rating / 10, 0), 1)
20
    except:
21
        return 0.5  # 默认中等重要性
22

23

24
# 示例
25
events = [
26
    "John 在咖啡厅买了一杯拿铁",           # 重要性 ~0.2
27
    "John 和 Maria 开始约会",               # 重要性 ~0.8
28
    "John 决定参加市议会选举",              # 重要性 ~0.9
29
    "John 刷了牙",                          # 重要性 ~0.1
30
]
31

32
for event in events:
33
    importance = assess_importance(llm, event)
34
    print(f"事件：{event}")
35
    print(f"重要性：{importance:.2f}\n")

1
┌─────────────────────────────────────────────────────────────┐
2
│                    反思触发条件                               │
3
├─────────────────────────────────────────────────────────────┤
4
│                                                             │
5
│  条件 1：累计记忆达到阈值                                     │
6
│     - 当 Agent 累积了 100 条新记忆                           │
7
│     - 触发一次反思                                           │
8
│                                                             │
9
│  条件 2：时间间隔                                            │
10
│     - 每隔一定时间（如虚拟世界的几小时）                       │
11
│     - 触发反思                                               │
12
│                                                             │
13
│  反思结果存储到 Memory Stream                                 │
14
│     - 作为特殊记忆类型                                        │
15
│     - 重要性通常较高                                          │
16
│                                                             │
17
└─────────────────────────────────────────────────────────────┘

1
from typing import List, Tuple
2

3
class ReflectionEngine:
4
    """反思引擎"""
5

6
    def __init__(self, llm, memory_stream: MemoryStream):
7
        self.llm = llm
8
        self.memory_stream = memory_stream
9

10
    def reflect(self, agent_name: str) -> List[str]:
11
        """执行反思过程"""
12
        # 步骤 1：检索最近的记忆
13
        recent_memories = self.memory_stream.retrieve_recent(n=100)
14

15
        # 步骤 2：生成反思问题
16
        questions = self._generate_questions(recent_memories)
17

18
        # 步骤 3：对每个问题生成洞察
19
        insights = []
20
        for question in questions:
21
            # 检索与问题相关的记忆
22
            relevant_memories = self.memory_stream.retrieve(question, top_k=20)
23

24
            # 生成洞察
25
            insight = self._generate_insight(question, relevant_memories)
26
            insights.append(insight)
27

28
            # 存储洞察到记忆流
29
            self.memory_stream.add_memory(
30
                content=f"[Reflection] {insight}",
31
                importance=0.9  # 反思结果重要性高
32
            )
33

34
        return insights
35

36
    def _generate_questions(self, memories: List[Memory]) -> List[str]:
37
        """基于记忆生成反思问题"""
38
        memory_texts = "\n".join([f"- {m.content}" for m in memories])
39

40
        prompt = f"""
41
        Given only the information above, what are 3 most
42
        salient high-level questions we can answer about the
43
        subjects in the statements?
44

45
        Statements:
46
        {memory_texts}
47

48
        Questions (1, 2, 3):
49
        """
50

51
        response = self.llm.generate(prompt)
52
        # 解析问题
53
        questions = self._parse_questions(response)
54
        return questions
55

56
    def _generate_insight(self, question: str,
57
                         memories: List[Memory]) -> str:
58
        """针对问题生成洞察"""
59
        memory_texts = "\n".join([f"- {m.content}" for m in memories])
60

61
        prompt = f"""
62
        Statements about John:
63
        {memory_texts}
64

65
        Question: {question}
66

67
        What is a high-level insight or inference you can draw
68
        from these statements in 2 sentences or less?
69
        """
70

71
        insight = self.llm.generate(prompt)
72
        return insight.strip()
73

74
    def _parse_questions(self, response: str) -> List[str]:
75
        """解析 LLM 返回的问题列表"""
76
        questions = []
77
        for line in response.strip().split("\n"):
78
            # 移除编号
79
            line = line.strip()
80
            if line and line[0].isdigit():
81
                line = line.split(".", 1)[-1].strip()
82
            if line:
83
                questions.append(line)
84
        return questions[:3]  # 最多 3 个问题

1
Agent：John
2

3
=== 最近记忆 ===
4
- John 在图书馆看到了 Maria
5
- John 主动和 Maria 搭话
6
- John 问 Maria 是否愿意一起学习
7
- John 送了 Maria 一本书
8
- John 在提到 Maria 时会微笑
9

10
=== 生成的反思问题 ===
11
1. John 对 Maria 的态度是什么？
12
2. John 最近的行为有什么共同点？
13
3. John 和 Maria 的关系正在如何发展？
14

15
=== 生成的洞察 ===
16
1. John 对 Maria 有明显的好感，他主动接近她并创造相处机会。
17
2. John 正在尝试与 Maria 建立更亲密的关系。
18
3. John 可能正在发展一段浪漫关系。
19

20
=== 存储到记忆流 ===
21
[Reflection] John 对 Maria 有好感，正在主动追求她。

1
from dataclasses import dataclass
2
from typing import List, Optional
3
from datetime import datetime, timedelta
4

5
@dataclass
6
class Plan:
7
    """计划"""
8
    description: str           # 计划描述
9
    start_time: datetime       # 开始时间
10
    end_time: datetime         # 结束时间
11
    sub_plans: List['Plan']    # 子计划
12
    status: str = "pending"    # pending/ongoing/completed
13

14
class PlanningEngine:
15
    """规划引擎"""
16

17
    def __init__(self, llm, memory_stream: MemoryStream):
18
        self.llm = llm
19
        self.memory_stream = memory_stream
20
        self.current_plan: Optional[Plan] = None
21

22
    def create_daily_plan(self, agent_name: str,
23
                         current_time: datetime) -> Plan:
24
        """创建日计划"""
25
        # 检索相关记忆和反思
26
        memories = self.memory_stream.retrieve(
27
            f"What should {agent_name} do today?",
28
            top_k=20
29
        )
30

31
        # 检索反思结果
32
        reflections = [
33
            m for m in self.memory_stream.memories
34
            if "[Reflection]" in m.content
35
        ][-5:]  # 最近 5 条反思
36

37
        # 构建提示词
38
        prompt = self._build_plan_prompt(
39
            agent_name, current_time, memories, reflections
40
        )
41

42
        # 生成计划
43
        response = self.llm.generate(prompt)
44

45
        # 解析计划
46
        daily_plan = self._parse_plan(response, current_time)
47
        self.current_plan = daily_plan
48

49
        return daily_plan
50

51
    def _build_plan_prompt(self, agent_name: str,
52
                          current_time: datetime,
53
                          memories: List[Memory],
54
                          reflections: List[Memory]) -> str:
55
        """构建规划提示词"""
56
        memory_texts = "\n".join([f"- {m.content}" for m in memories])
57
        reflection_texts = "\n".join([f"- {m.content}" for m in reflections])
58

59
        prompt = f"""
60
        Name: {agent_name}
61
        Current time: {current_time.strftime("%Y-%m-%d %H:%M")}
62

63
        Recent memories:
64
        {memory_texts}
65

66
        Reflections about self:
67
        {reflection_texts}
68

69
        Today is {current_time.strftime("%A, %B %d, %Y")}.
70
        Here is {agent_name}'s plan for the day:
71

72
        1. [Time] Activity
73
        2. [Time] Activity
74
        ...
75

76
        Please create a detailed schedule for the entire day.
77
        """
78
        return prompt
79

80
    def refine_plan(self, plan: Plan, current_time: datetime) -> Plan:
81
        """细化计划到小时级别"""
82
        prompt = f"""
83
        Current plan: {plan.description}
84
        Current time: {current_time.strftime("%H:%M")}
85

86
        Please refine this plan into specific 1-hour segments:
87

88
        Hour | Activity
89
        ---- | --------
90
        {current_time.hour}:00 |
91
        {current_time.hour + 1}:00 |
92
        ...
93
        """
94

95
        response = self.llm.generate(prompt)
96
        refined_plan = self._parse_hourly_plan(response, current_time)
97
        return refined_plan
98

99
    def get_next_action(self, agent_name: str,
100
                        current_time: datetime) -> str:
101
        """获取下一个具体行动"""
102
        if self.current_plan is None:
103
            self.create_daily_plan(agent_name, current_time)
104

105
        # 找到当前应该执行的计划
106
        current_activity = self._find_current_activity(
107
            self.current_plan, current_time
108
        )
109

110
        if current_activity is None:
111
            return "idle"
112

113
        # 生成具体行动
114
        prompt = f"""
115
        Current activity: {current_activity.description}
116
        Current location: {current_activity.location if hasattr(current_activity, 'location') else 'home'}
117

118
        What is the immediate next action {agent_name} should take?
119
        Answer in one sentence, starting with an action verb.
120
        """
121

122
        action = self.llm.generate(prompt)
123
        return action.strip()
124

125
    def _find_current_activity(self, plan: Plan,
126
                               current_time: datetime) -> Optional[Plan]:
127
        """找到当前应该执行的活动"""
128
        for sub_plan in plan.sub_plans:
129
            if sub_plan.start_time <= current_time < sub_plan.end_time:
130
                return sub_plan
131
        return None
132

133
    def _parse_plan(self, response: str,
134
                   current_time: datetime) -> Plan:
135
        """解析 LLM 返回的计划"""
136
        lines = response.strip().split("\n")
137
        sub_plans = []
138

139
        for line in lines:
140
            if line.strip() and line[0].isdigit():
141
                # 解析格式如 "1. [09:00] Morning routine"
142
                parts = line.split("]", 1)
143
                if len(parts) == 2:
144
                    time_str = parts[0].split("[")[-1]
145
                    activity = parts[1].strip()
146

147
                    # 创建子计划
148
                    hour = int(time_str.split(":")[0])
149
                    start = current_time.replace(hour=hour, minute=0)
150
                    end = start + timedelta(hours=1)
151

152
                    sub_plans.append(Plan(
153
                        description=activity,
154
                        start_time=start,
155
                        end_time=end,
156
                        sub_plans=[]
157
                    ))
158

159
        return Plan(
160
            description=f"Daily plan for {current_time.strftime('%Y-%m-%d')}",
161
            start_time=current_time.replace(hour=0, minute=0),
162
            end_time=current_time.replace(hour=23, minute=59),
163
            sub_plans=sub_plans
164
        )
165

166
    def _parse_hourly_plan(self, response: str,
167
                          current_time: datetime) -> Plan:
168
        """解析小时级别计划"""
169
        # 类似 _parse_plan 的实现
170
        pass

1
Agent：John
2
日期：2024-03-15
3

4
=== 日计划 ===
5
09:00 - 起床，洗漱
6
10:00 - 去图书馆学习
7
12:00 - 和 Maria 一起午餐
8
13:00 - 下午课程
9
15:00 - 继续在图书馆研究
10
17:00 - 去 Hobbs Cafe 喝咖啡
11
18:00 - 准备晚餐
12
19:00 - 晚间阅读
13
21:00 - 休息
14

15
=== 当前时间 14:30 ===
16
当前活动：下午课程（13:00-15:00）
17
下一步行动：继续听课
18

19
=== 当前时间 15:00 ===
20
当前活动：继续在图书馆研究（15:00-17:00）
21
下一步行动：走向图书馆

1
@dataclass
2
class AgentProfile:
3
    """Agent 初始设定"""
4
    name: str
5
    age: int
6
    occupation: str
7
    personality: List[str]     # 性格特征
8
    background: str            # 背景故事
9
    relationships: Dict[str, str]  # 与其他 Agent 的关系
10
    current_location: str
11

12
# 示例 Agent
13
john_profile = AgentProfile(
14
    name="John Lin",
15
    age=25,
16
    occupation="PhD student in computer science",
17
    personality=["friendly", "curious", "hardworking"],
18
    background="""
19
    John Lin is a PhD student in computer science at
20
    Stanford University. He is passionate about AI
21
    research. He lives in a dorm with his roommate Sam.
22
    """,
23
    relationships={
24
        "Sam Moore": "roommate and close friend",
25
        "Maria Lopez": "lab partner, potential romantic interest"
26
    },
27
    current_location="dorm"
28
)

1
移除记忆流：Agent 行为失去连贯性，重复同样问题
2
移除反思：Agent 缺乏长期目标，行为短视
3
移除规划：Agent 行为随机，没有日程感
4

5
完整系统表现最佳，三个组件相互配合

1
每次 Agent 行动需要的 LLM 调用：
2
├── 感知：1 次调用（提取事件）
3
├── 记忆检索：1-2 次调用（生成嵌入、排序）
4
├── 规划：1-3 次调用（生成计划、细化）
5
├── 行动：1 次调用
6
└── 反思（定期）：3-5 次调用
7

8
25 个 Agent × 每天约 100 次行动 × 多次调用
9
= 大量 API 成本

1
# 当前实现的问题
2
class MemoryStream:
3
    def retrieve(self, query: str, top_k: int = 10):
4
        # 线性扫描所有记忆
5
        for memory in self.memories:  # O(n) 复杂度
6
            score = self._calculate_score(query, memory)
7
            ...
8

9
# 问题：随着时间推移，记忆数量持续增长
10
# 长期运行后检索变慢

1
计划执行中的问题：
2
┌─────────────────────────────────────────────────────────────┐
3
│                                                             │
4
│  Agent 制定了计划，但环境可能变化：                           │
5
│                                                             │
6
│  计划：下午 3 点去图书馆                                      │
7
│  变化：图书馆临时关闭                                        │
8
│  问题：Agent 如何动态调整？                                  │
9
│                                                             │
10
│  论文的解决方案有限，主要依赖重新规划                          │
11
│  缺乏细粒度的反应式调整机制                                   │
12
│                                                             │
13
└─────────────────────────────────────────────────────────────┘

1
1. 记忆遗忘机制
2
   - 记忆只会增加，不会遗忘
3
   - 不符合人类记忆特点
4
   - 长期运行后记忆库膨胀
5

6
2. 情感建模
7
   - Agent 的情感表达较为表面
8
   - 缺乏深层的情感状态变化
9
   - 情感对决策的影响有限
10

11
3. 学习能力
12
   - Agent 不会真正「学习」新技能
13
   - 知识来源于 LLM，不会积累
14
   - 无法超出 LLM 的能力边界
15

16
4. 社会规范
17
   - Agent 对社会规范的理解有限
18
   - 可能产生不符合伦理的行为
19
   - 缺乏价值观引导机制

1
# Generative Agents 风格
2
agent = GenerativeAgent(
3
    name="John",
4
    profile=john_profile,
5
    memory_stream=MemoryStream(),
6
    reflection_engine=ReflectionEngine(),
7
    planning_engine=PlanningEngine()
8
)
9
agent.live()  # Agent 自主生活
10

11
# CrewAI 风格
12
agent = Agent(
13
    role="Researcher",
14
    goal="Research AI topics",
15
    backstory="Expert in AI research"
16
)
17
task = Task(description="...", agent=agent)
18
crew = Crew(agents=[agent], tasks=[task])
19
crew.kickoff()  # 执行特定任务

1
┌─────────────────────────────────────────────────────────────┐
2
│                 论文启发的现代设计                            │
3
├─────────────────────────────────────────────────────────────┤
4
│                                                             │
5
│  1. 记忆系统成为标配                                         │
6
│     ├── MemGPT：显式的记忆管理架构                           │
7
│     ├── LangChain Memory：多种记忆类型                       │
8
│     └── 向量数据库：长期记忆的标准实现                        │
9
│                                                             │
10
│  2. 反思机制被广泛采用                                       │
11
│     ├── Reflexion 论文：将反思系统化                         │
12
│     ├── Self-RAG：检索增强的反思                             │
13
│     └── 多轮对话中的自我修正                                 │
14
│                                                             │
15
│  3. 规划系统的发展                                           │
16
│     ├── Plan-and-Execute：显式规划架构                       │
17
│     ├── ReAct：思考-行动循环                                 │
18
│     └── Tree of Thoughts：多路径规划                         │
19
│                                                             │
20
│  4. Multi-Agent 系统                                         │
21
│     ├── AutoGen：微软的多 Agent 框架                         │
22
│     ├── CrewAI：角色扮演协作                                 │
23
│     └── MetaGPT：软件公司模拟                                │
24
│                                                             │
25
└─────────────────────────────────────────────────────────────┘