AGI大模型应用开发实战：构建具备长期记忆与自主反思的智能助手

原创

用户12687280

发布于 2026-08-29 18:16:07

700

当前大模型应用正从"单轮问答"迈向"类AGI认知体"——具备持续记忆、多模态感知、自我反思与工具调用能力的智能系统。本文将系统性地构建一个 "MemAgent" ，它不再是被动响应工具，而是能够主动回忆过往对话、从交互中归纳知识、通过视觉理解环境，并调用外部工具完成复杂任务。我们将深入拆解其 记忆流架构、反思机制 与 多模态工具链，附全部核心代码。

一、认知架构设计：记忆、反思与行动的闭环

MemAgent 基于 认知心理学 的"工作记忆-长期记忆-反思"三层模型设计：

工作记忆：当前对话上下文（滑动窗口，保留最近20轮）。
长期记忆：所有历史交互经Embedding后存入向量数据库，支持语义检索。
反思机制：每N轮交互触发一次"自我总结"，将零散事实归纳为结构化知识（如"用户偏好编程"、"反感广告"），存入独立的知识图谱。
行动模块：多模态感知（图像/音频）结合工具调用（搜索、计算、绘图）。

全流程由 LangGraph 编排状态流转，保证循环可控。

二、长期记忆系统：层次化记忆存储与检索

我们采用 分层记忆 策略：短期记忆存Redis（TTL=1h），长期记忆存 Chroma 向量库，反思知识存 Neo4j 图数据库。

import chromadb
from sentence_transformers import SentenceTransformer
from datetime import datetime
import json

embedder = SentenceTransformer('BAAI/bge-large-zh-v1.5')
chroma_client = chromadb.PersistentClient(path="./memory_db")
collection = chroma_client.get_or_create_collection(name="episodic_memory")

class MemorySystem:
    def add(self, text: str, metadata: dict):
        """存入长期记忆，附带时间戳和重要性评分"""
        emb = embedder.encode(text).tolist()
        doc_id = f"mem_{datetime.now().timestamp()}"
        collection.add(
            documents=[text],
            embeddings=[emb],
            metadatas=[{**metadata, "timestamp": datetime.now().isoformat()}],
            ids=[doc_id]
        )
        return doc_id
    
    def retrieve(self, query: str, top_k=5) -> list:
        """语义检索相关记忆"""
        q_emb = embedder.encode(query).tolist()
        results = collection.query(query_embeddings=[q_emb], n_results=top_k)
        return [{"text": doc, "meta": meta} for doc, meta in zip(results['documents'][0], results['metadatas'][0])]
    
    def forget_old(self, days=7):
        """定期遗忘：删除7天前的记忆（模拟人类遗忘曲线）"""
        # 实现略，按timestamp过滤删除
        pass

三、反思机制：从记忆到知识的自我归纳

反思是AGI系统从"记忆"进化到"理解"的关键。我们设计一个反思节点，在对话步数达到阈值时自动触发，调用LLM对近期记忆进行摘要和规律提取：

from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate

reflection_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)

def reflect_on_memories(recent_interactions: list[str]) -> dict:
    """分析最近10轮对话，提取用户偏好和意图模式"""
    prompt = PromptTemplate.from_template("""
    分析以下用户与助手的交互记录，提取：
    1. 用户的长期偏好（3个，如：喜欢技术、厌恶广告、偏好简洁）
    2. 用户的情绪模式（积极/消极/中性）
    3. 可能的目标意图（如：学习某技术、解决某问题）
    
    交互记录：
    {interactions}
    
    输出JSON格式：{{"preferences": [...], "emotion": "...", "goals": [...]}}
    """)
    chain = prompt | reflection_llm
    response = chain.invoke({"interactions": "\n".join(recent_interactions[-10:])})
    # 解析JSON并存入知识图谱（Neo4j）
    return json.loads(response.content)

反思结果存入 知识图谱，后续检索时优先匹配，大幅提升回答个性化程度。

四、多模态感知模块：让AGI看懂世界

AGI需要理解图像和音频。我们集成 CLIP 做图像语义理解， Whisper 做语音转文字，并将视觉特征作为记忆的一部分存储。

import torch
from PIL import Image
from transformers import CLIPProcessor, CLIPModel, WhisperProcessor, WhisperForConditionalGeneration

clip_model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32").to("cuda")
clip_processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")

def image_to_memory(image_path: str) -> str:
    """将图像转换为文本描述并存入记忆"""
    image = Image.open(image_path)
    inputs = clip_processor(images=image, return_tensors="pt").to("cuda")
    # 使用CLIP生成图像嵌入，配合预定义候选标签生成描述
    candidate_texts = ["a person", "a dog", "a landscape", "food", "technology", "document"]
    text_inputs = clip_processor(text=candidate_texts, return_tensors="pt", padding=True).to("cuda")
    with torch.no_grad():
        image_features = clip_model.get_image_features(**inputs)
        text_features = clip_model.get_text_features(**text_inputs)
        similarity = (image_features @ text_features.T).softmax(dim=-1)
    best_idx = similarity.argmax().item()
    description = f"图像内容: {candidate_texts[best_idx]}, 置信度: {similarity[0][best_idx].item():.2f}"
    return description

# 语音转文字（使用Whisper）
whisper_model = WhisperForConditionalGeneration.from_pretrained("openai/whisper-base")
whisper_processor = WhisperProcessor.from_pretrained("openai/whisper-base")

def audio_to_text(audio_path: str) -> str:
    inputs = whisper_processor(Image.open(audio_path), return_tensors="pt").input_features
    generated_ids = whisper_model.generate(inputs)
    return whisper_processor.batch_decode(generated_ids, skip_special_tokens=True)[0]

多模态输入被统一转化为文本描述，存入记忆系统，让AGI能够"回忆"之前见过的图像或听到的对话。

五、动态工具编排：基于意图的智能调用

AGI需要根据用户意图自动选择工具。我们实现一个 意图分类器 + 工具注册表，动态构建调用链：

from pydantic import BaseModel
from typing import List, Dict
import requests

class Tool:
    def __init__(self, name: str, func, description: str, schema: Dict):
        self.name = name
        self.func = func
        self.description = description
        self.schema = schema

# 注册工具
tools = [
    Tool("web_search", 
         lambda q: requests.get(f"https://api.tavily.com/search?q={q}").json(),
         "搜索互联网信息", 
         {"query": "string"}),
    Tool("calculate",
         lambda expr: eval(expr),
         "执行数学计算",
         {"expression": "string"}),
    Tool("generate_image",
         lambda prompt: "模拟生成图片",  # 实际调用Stable Diffusion
         "根据文本生成图像",
         {"prompt": "string"})
]

def autoselect_tools(user_query: str, memory_context: str) -> list[str]:
    """基于用户查询和记忆，决定调用哪些工具"""
    prompt = f"""用户问题：{user_query}
    历史记忆：{memory_context}
    可用工具：{[t.name + ":" + t.description for t in tools]}
    请返回需要调用的工具名称列表（JSON数组），若无需工具返回[]。"""
    response = reflection_llm.invoke(prompt)
    return json.loads(response.content)

工具调用结果与反思记忆结合，形成完整的认知-行动闭环。

六、全链路编排：LangGraph状态机实现

我们将上述模块整合为 状态图，确保AGI循环可控：

from langgraph.graph import StateGraph, END
from typing import TypedDict

class AgentState(TypedDict):
    user_input: str
    memory_context: list
    reflections: dict
    tool_results: list
    final_response: str
    step: int

def perceive_node(state: AgentState):
    """多模态感知（如果是图像/音频则转文本）"""
    # 检测输入类型，调用相应函数
    return state

def recall_node(state: AgentState):
    """检索长期记忆"""
    mems = memory_system.retrieve(state['user_input'], top_k=3)
    return {**state, "memory_context": mems}

def reflect_node(state: AgentState):
    """每5步触发一次反思"""
    if state['step'] % 5 == 0:
        ref = reflect_on_memories(state.get('history', []))
        return {**state, "reflections": ref}
    return state

def act_node(state: AgentState):
    """选择并执行工具"""
    tools_to_call = autoselect_tools(state['user_input'], str(state['memory_context']))
    results = []
    for name in tools_to_call:
        tool = next(t for t in tools if t.name == name)
        res = tool.func(state['user_input'])
        results.append({"tool": name, "result": res})
    return {**state, "tool_results": results}

def generate_node(state: AgentState):
    """生成最终回答（融合记忆+反思+工具结果）"""
    context = f"记忆：{state['memory_context']}\n反思：{state.get('reflections', {})}\n工具结果：{state['tool_results']}"
    response = llm.invoke(f"{context}\n\n用户：{state['user_input']}\n助手：")
    return {**state, "final_response": response.content}

# 构建图
graph = StateGraph(AgentState)
graph.add_node("perceive", perceive_node)
graph.add_node("recall", recall_node)
graph.add_node("reflect", reflect_node)
graph.add_node("act", act_node)
graph.add_node("generate", generate_node)

graph.set_entry_point("perceive")
graph.add_edge("perceive", "recall")
graph.add_edge("recall", "reflect")
graph.add_edge("reflect", "act")
graph.add_edge("act", "generate")
graph.add_edge("generate", END)

app = graph.compile()

七、服务化与生产级优化

最后用 FastAPI 暴露接口，并加入以下优化：

from fastapi import FastAPI, UploadFile
app = FastAPI()

@app.post("/chat")
async def chat(input_text: str, file: UploadFile = None):
    # 如果上传图片，先存入记忆
    if file:
        img_desc = image_to_memory(await file.read())
        memory_system.add(img_desc, {"type": "image"})
    
    result = app.invoke({
        "user_input": input_text,
        "memory_context": [],
        "reflections": {},
        "tool_results": [],
        "final_response": "",
        "step": 0
    })
    return {"response": result['final_response']}

生产优化：

记忆压缩：超过1000条记忆时，触发层次聚类合并相似条目。
反思频率自适应：根据对话复杂度动态调整反思间隔（简单问答每10步，复杂推理每3步）。
工具调用超时控制：设置asyncio.timeout，超时则返回工具不可用提示。
流式输出：使用SSE（Server-Sent Events）逐词返回，提升体验。

结语

本文完整构建了一个具备 长期记忆、自主反思、多模态感知与工具编排 的AGI应用系统，核心突破了"被动响应"的局限，向"主动认知"迈进了一大步。这套架构的可扩展性极强：记忆层可替换为更高效的HNSW索引，反思机制可升级为多智能体辩论（Multi-Agent Debate），工具链可接入代码执行、浏览器操作等复杂能力。AGI大模型应用开发的本质，不再是调用API，而是 设计认知流——让模型在记忆与反思中持续进化，真正向通用智能靠近一步。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

大模型部署

agile