从理论到实践：深度解析Agent记忆系统与ReAct框架，附小红书爆款案例

class User(Base):
    id = Column(Integer, primary_key=True)
    username = Column(String(100), unique=True, nullable=False)
    email = Column(String(255), unique=True, nullable=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow)

class Session(Base):
    id = Column(Integer, primary_key=True)
    session_id = Column(String(100), unique=True, nullable=False)
    user_id = Column(Integer, ForeignKey('users.id'))
    title = Column(String(255), nullable=True)
    context_window = Column(Integer, default=10)  # 上下文窗口大小
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)

class Memory(Base):
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'))
    session_id = Column(Integer, ForeignKey('sessions.id'), nullable=True)

    memory_type = Column(String(50))  # episodic/semantic/procedural
    content = Column(Text, nullable=False)
    metadata = Column(JSON, nullable=True)

    vector_id = Column(String(100))  # 关联ChromaDB的向量ID
    importance_score = Column(Float, default=0.5)  # 重要性评分

    access_count = Column(Integer, default=0)  # 被访问次数
    last_accessed_at = Column(DateTime, nullable=True)
    is_deleted = Column(Boolean, default=False)  # 软删除标记

class VectorStore:
    def __init__(self, persist_directory: str = "./chat_chroma.db"):
        self.client = chromadb.Client(Settings(
            persist_directory=persist_directory,
            anonymized_telemetry=False
        ))
        self.collection = self.client.get_or_create_collection(
            name="memory_collection"
        )

    def add_memory(self, content: str, metadata: Dict = None) -> str:
        """添加记忆到向量库"""
        memory_id = str(uuid.uuid4())
        metadata = metadata or {}
        metadata['created_at'] = datetime.utcnow().isoformat()

        self.collection.add(
            ids=[memory_id],
            documents=[content],
            metadatas=[metadata]
        )
        return memory_id

    def search_memories(self, query: str, n_results: int = 5,
                       filter_metadata: Dict = None) -> Dict:
        """语义搜索记忆"""
        results = self.collection.query(
            query_texts=[query],
            n_results=n_results,
            where=filter_metadata
        )
        return {
            'ids': results['ids'][0],
            'documents': results['documents'][0],
            'metadatas': results['metadatas'][0],
            'distances': results['distances'][0]
        }

def add_memory(self, user_id: int, content: str,
               memory_type: str = "episodic",
               session_id: int = None,
               metadata: Dict = None,
               importance_score: float = 0.5) -> Memory:
    """添加记忆（同时保存到数据库和向量库）"""
    db = self._get_db_session()
    try:
        # 1. 创建数据库记录
        memory = Memory(
            user_id=user_id,
            session_id=session_id,
            memory_type=memory_type,
            content=content,
            metadata=metadata or {},
            importance_score=importance_score
        )
        db.add(memory)
        db.commit()
        db.refresh(memory)

        # 2. 添加到向量库
        vector_metadata = {
            'user_id': user_id,
            'memory_type': memory_type,
            'memory_db_id': memory.id
        }
        vector_id = self.vector_store.add_memory(
            content=content,
            metadata=vector_metadata
        )

        # 3. 更新向量ID关联
        memory.vector_id = vector_id
        db.commit()

        return memory
    finally:
        db.close()

def search_memories(self, query: str, user_id: int = None,
                   memory_type: str = None,
                   n_results: int = 5) -> List[Dict]:
    """搜索记忆"""
    # 1. 在向量库中进行语义搜索
    filter_metadata = {}
    if user_id:
        filter_metadata['user_id'] = user_id
    if memory_type:
        filter_metadata['memory_type'] = memory_type

    results = self.vector_store.search_memories(
        query=query,
        n_results=n_results * 2,  # 多检索一些，用于后续过滤
        filter_metadata=filter_metadata if filter_metadata else None
    )

    # 2. 整合数据库中的详细信息
    memories = []
    db = self._get_db_session()
    try:
        for i, vector_id in enumerate(results['ids']):
            metadata = results['metadatas'][i]
            memory_db_id = metadata.get('memory_db_id')

            if memory_db_id:
                memory = db.query(Memory).filter(
                    Memory.id == memory_db_id,
                    Memory.is_deleted == False
                ).first()

                if memory:
                    # 更新访问统计信息
                    memory.access_count += 1
                    memory.last_accessed_at = datetime.utcnow()

                    memories.append({
                        'id': memory.id,
                        'content': memory.content,
                        'memory_type': memory.memory_type,
                        'importance_score': memory.importance_score,
                        'distance': results['distances'][i],
                        'created_at': memory.created_at.isoformat()
                    })

            if len(memories) >= n_results:
                break

        db.commit()
        return memories
    finally:
        db.close()

def retrieve_for_query(self, user_id: int, query: str,
                      session_id: int = None,
                      include_context_window: bool = True,
                      max_results: int = 5) -> Dict:
    """RAG检索：整合短期和长期记忆"""
    result = {
        'context_window': [],
        'relevant_memories': []
    }

    # 1. 获取上下文窗口（短期记忆）
    if include_context_window and session_id:
        context_memories = self.service.get_context_window(session_id)
        result['context_window'] = [
            {'content': m.content, 'created_at': m.created_at.isoformat()}
            for m in context_memories
        ]

    # 2. 语义搜索（长期记忆）
    relevant = self.service.search_memories(
        query=query,
        user_id=user_id,
        n_results=max_results
    )
    result['relevant_memories'] = relevant
    return result

class ChatWithMemory:
    def chat(self, user_input: str, use_rag: bool = True) -> str:
        """与LLM进行带记忆的对话"""
        messages = [{"role": "system", "content": self.system_prompt}]

        if use_rag:
            # 执行RAG检索
            rag_result = self.memory.retrieve_for_query(
                user_id=self.user_id,
                query=user_input,
                session_id=self.session_id,
                include_context_window=True,
                max_results=3
            )

            # 添加相关的长期记忆到上下文
            if rag_result['relevant_memories']:
                memory_context = "相关历史信息：\n"
                for mem in rag_result['relevant_memories'][:3]:
                    memory_context += f"- {mem['content']}\n"
                messages.append({"role": "system", "content": memory_context})

            # 添加上下文窗口中的短期记忆
            for ctx in rag_result['context_window']:
                # 解析对话内容（假设格式为"User: ...\nAssistant: ..."）
                if "User:" in ctx['content'] and "Assistant:" in ctx['content']:
                    parts = ctx['content'].split("\nAssistant:")
                    user_part = parts[0].replace("User:", "").strip()
                    assistant_part = parts[1].strip()

                    messages.append({"role": "user", "content": user_part})
                    messages.append({"role": "assistant", "content": assistant_part})

        # 添加当前用户输入
        messages.append({"role": "user", "content": user_input})

        # 调用LLM
        response = call_llm(messages=messages)
        assistant_message = response.choices[0].message.content

        # 保存本轮对话到记忆系统
        self.memory.add_conversation_memory(
            user_id=self.user_id,
            session_id=self.session_id,
            user_message=user_input,
            assistant_message=assistant_message
        )

        return assistant_message

# 基础使用
from memory_tools import MemoryTools

# 初始化
memory = MemoryTools()

# 创建用户和会话
user_id = memory.create_user("alice")
session_id = memory.create_session(user_id, "session-001")

# 添加对话记忆
memory.add_conversation_memory(
    user_id=user_id,
    session_id=session_id,
    user_message="我喜欢Python编程",
    assistant_message="Python是一门很棒的语言！"
)

# 搜索记忆
results = memory.search_memories(
    user_id=user_id,
    query="编程语言",
    n_results=5
)

# 运行交互式聊天程序
python chat_with_memory.py

# 示例对话
你: 我叫Alice，喜欢编程
助手: 你好Alice！很高兴认识你。编程是一个很有趣的领域...

你: 我之前说过我叫什么名字？
助手: 你之前说你叫Alice。

@dataclass
class PlanStep:
    id: str
    tool: str
    args: Dict[str, Any]
    why: str = ""
    status: Literal["pending", "running", "success", "failed", "skipped"] = "pending"
    output: Any = None
    error: str = ""

@dataclass
class Plan:
    goal: str
    steps: List[PlanStep] = field(default_factory=list)

system_prompt = f"""
你是一个任务规划器。你必须只输出JSON，不要输出任何解释文字。
目标：将用户的需求拆解为若干个步骤，每一步需要调用一个指定的工具，并提供相应的参数。

关键规则——数据依赖：
如果某一步的参数需要依赖前面步骤的输出结果，请务必使用模板语法 `{{{{steps.前序步骤ID.output.字段名}}}}`。
不要自己捏造数据，而是严格引用前序步骤的实际输出。

例如：
1. step1 (id="s1") 调用 geocode 工具，返回 `{{"location": "116.40,39.90"}}`
2. step2 (id="s2") 需要用到这个坐标，其参数应写为： `{{"location": "{{{{steps.s1.output.location}}}}"}}`

你必须输出的JSON结构：
{{
  "goal": "...",
  "steps": [
    {{
      "id": "step1",
      "tool": "tool_name",
      "args": {{ ... }}
    }}
  ]
}}
"""

async def _run_step(self, step: PlanStep, mcp_client: Client, context: Dict[str, Any]) -> None:
    step.status = "running"
    try:
        # 解析参数中的 {{...}} 模板，替换为上下文中的真实数据
        resolved_args = resolve_templates(step.args, context)

        # 通过MCP客户端调用指定的工具
        result = await mcp_client.call_tool(step.tool, resolved_args)

        # 记录工具执行的输出结果
        step.output = getattr(result, "structured_content", None) or result.content
        step.status = "success"

        # 将本步骤的输出写入上下文，供后续步骤引用
        context["steps"][step.id] = {
            "status": step.status,
            "output": step.output
        }
    except Exception as e:
        step.status = "failed"
        step.error = str(e)

async function main({ params }: Args): Promise<Output> {
    // 构建输出对象
   const ret = {
    fields: [
        {
            fields: {
                "小红书文案": params.content,
                "小红书标题": params.title,
                "小红书图文地址": params.imageUrl

            },
            record_id: params.record_id
        }
    ]
};
    return ret;
}