基于Python实现一个具有记忆功能的AI对话代理

import os
from dotenv import load_dotenv
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.db.sqlite import SqliteDb

load_dotenv(".env")
user_id = "ava"

db = SqliteDb(
    db_file="mem.db",
    memory_table="user_memories",  # 为记忆数据指定表名
)

memory_agent = Agent(
    model=OpenAIChat(
        id="qwen/qwen3-4b-thinking-2507", 
        base_url=os.getenv("BASE_URL"), 
        api_key=os.getenv("OPENAI_API_KEY")
    ),
    db=db,
    enable_agentic_memory=True,
    enable_user_memories=True,
    markdown=True,
)

db.clear_memories()

while True:
    user_input = input("user:")
    print("agent:")
    memory_agent.print_response(
        user_input,
        user_id=user_id,
        stream=True,
    )

from agno.workflow import Workflow, Step, Parallel
from agno.agent import Agent

# 1. 定义多个专业化的智能体
web_researcher = Agent(
    name="网络研究员",
    role="从互联网搜索最新信息和新闻",
    # ... 配置工具和指令
)

academic_researcher = Agent(
    name="学术研究员",
    role="从学术数据库查找研究和论文",
    # ... 配置工具和指令
)

social_media_analyst = Agent(
    name="社交媒体分析师",
    role="分析社交媒体平台上的趋势和观点",
    # ... 配置工具和指令
)

# 2. 定义总结智能体
summarizer = Agent(
    name="首席分析官",
    role="整合多来源信息，生成一份全面、结构化的总结报告",
    instructions="综合所有研究员的发现，识别共同点和关键差异，形成权威结论。"
)

# 3. 构建工作流
research_workflow = Workflow(
    name="综合研究工作流",
    steps=[
        Parallel(  # 并行执行以下所有研究步骤

            [
                Step(name="网络研究", agent=web_researcher),
                Step(name="学术研究", agent=academic_researcher),
                Step(name="社媒分析", agent=social_media_analyst),
            ]
        ),
        Step(name="生成总结报告", agent=summarizer)  # 汇总并行结果
    ]
)

# 4. 运行工作流
response = research_workflow.run("关于神经形态计算的最新发展")