智能体(Agent)—— 从工具调用到自主任务执行

┌─────────────────────────────────────────────────────────┐│                      Agent 架构                          │├─────────────────────────────────────────────────────────┤│  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐   ││  │ Profile │  │ Memory  │  │Planning │  │ Action  │   ││  │  角色   │  │  记忆   │  │  规划   │  │  行动   │   ││  └─────────┘  └─────────┘  └─────────┘  └─────────┘   ││       │            │            │            │         ││       └────────────┴────────────┴────────────┘         ││                          │                              ││                    ┌─────▼─────┐                        ││                    │    LLM    │                        ││                    │  (大脑)   │                        ││                    └───────────┘                        │└─────────────────────────────────────────────────────────┘

Thought: 用户想知道今天的天气，我需要先获取位置信息Action: get_user_locationObservation: 北京Thought: 现在我知道了用户在北京，需要查询北京天气Action: get_weather("北京")Observation: 晴天，25°CAnswer: 北京今天天气晴朗，气温25°C

┌───────────────────────────────────────────────────┐│                  ReAct 循环                        ││                                                   ││   ┌─────────┐    ┌─────────┐    ┌─────────┐     ││   │ Thought │───▶│ Action  │───▶│Observation│    ││   │  思考   │    │  行动   │    │   观察   │     ││   └─────────┘    └─────────┘    └─────────┘     ││        ▲                               │         ││        └───────────────────────────────┘         ││                   (循环直到完成)                   │└───────────────────────────────────────────────────┘

# 规划阶段plan = """1. 收集用户需求文档2. 分析技术可行性3. 生成项目架构设计4. 编写代码框架5. 编写测试用例"""# 执行阶段for step in plan.steps:    result = execute_step(step)    if result.needs_revision:        revise_plan(result.feedback)

用户输入 ──▶ 规划器 ──▶ 任务列表 ──▶ 执行器 ──▶ 结果                           │                           ▼                      反馈循环

┌──────────────────────────────────────────────────┐│                  Multi-Agent 系统                 │├──────────────────────────────────────────────────┤│                                                  ││   ┌────────┐    ┌────────┐    ┌────────┐       ││   │ Leader │───▶│Analyst │    │ Coder  │       ││   │ Agent  │    │ Agent  │    │ Agent  │       ││   └────────┘    └────────┘    └────────┘       ││        │             │             │            ││        └─────────────┴─────────────┘            ││                      │                          ││              ┌───────▼───────┐                  ││              │ Shared Memory │                  ││              └───────────────┘                  │└──────────────────────────────────────────────────┘

research-agent/├── src/│   ├── agent/│   │   ├── __init__.py│   │   ├── planner.py      # 规划器│   │   ├── executor.py     # 执行器│   │   └── memory.py       # 记忆管理│   ├── tools/│   │   ├── arxiv.py        # arXiv 搜索│   │   ├── scholar.py      # Google Scholar│   │   └── summarizer.py   # 论文总结│   └── main.py├── requirements.txt└── README.md

# src/agent/planner.pyimport jsonfrom langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplateclass ResearchPlanner:    def __init__(self, model: str = "gpt-4"):        self.llm = ChatOpenAI(model=model)        self.prompt = ChatPromptTemplate.from_messages([            ("system", """你是一个研究规划专家。给定一个研究主题，你需要：1. 分解研究任务2. 确定需要搜索的关键词3. 规划研究步骤输出格式为 JSON：{  "tasks": [...],  "keywords": [...],  "steps": [...]}"""),            ("user", "{topic}")        ])    async def plan(self, topic: str) -> dict:        chain = self.prompt | self.llm        result = await chain.ainvoke({"topic": topic})        return json.loads(result.content)

# src/agent/executor.pyfrom typing import List, Dictfrom .memory import AgentMemoryclass ResearchExecutor:    def __init__(self, tools: List, memory: AgentMemory):        self.tools = {tool.name: tool for tool in tools}        self.memory = memory    async def execute(self, plan: dict) -> Dict:        results = {}        for step in plan["steps"]:            # 记录执行状态            self.memory.add_message("system", f"执行步骤: {step['description']}")            # 执行工具调用            if step["tool"] in self.tools:                tool = self.tools[step["tool"]]                result = await tool.arun(**step.get("params", {}))                results[step["id"]] = result                # 存储结果到记忆                self.memory.add_message("tool", result)        return results

# src/agent/memory.pyfrom typing import List, Dictfrom dataclasses import dataclass, fieldfrom datetime import datetime@dataclassclass Message:    role: str    content: str    timestamp: datetime = field(default_factory=datetime.now)class AgentMemory:    def __init__(self, max_short_term: int = 10):        self.short_term: List[Message] = []        self.long_term: List[Dict] = []  # 可接入向量数据库        self.max_short_term = max_short_term    def add_message(self, role: str, content: str):        message = Message(role=role, content=content)        self.short_term.append(message)        # 超过限制时，压缩历史        if len(self.short_term) > self.max_short_term:            self._compress_history()    def get_context(self) -> str:        """获取上下文用于 LLM 输入"""        return "\n".join([            f"{m.role}: {m.content}"            for m in self.short_term        ])    def _compress_history(self):        """压缩历史记忆"""        # 简单实现：保留最近的消息        compressed = self.short_term[:-self.max_short_term]        self.long_term.extend([{            "role": m.role,            "content": m.content,            "timestamp": m.timestamp.isoformat()        } for m in compressed])        self.short_term = self.short_term[-self.max_short_term:]

# src/tools/arxiv.pyfrom langchain.tools import BaseToolfrom pydantic import BaseModel, Fieldimport arxivclass ArxivSearchInput(BaseModel):    query: str = Field(description="搜索关键词")    max_results: int = Field(default=5, description="最大结果数")class ArxivSearchTool(BaseTool):    name: str = "arxiv_search"    description: str = "在 arXiv 上搜索学术论文"    args_schema: type[BaseModel] = ArxivSearchInput    def _run(self, query: str, max_results: int = 5) -> str:        search = arxiv.Search(            query=query,            max_results=max_results,            sort_by=arxiv.SortCriterion.relevance        )        results = []        for paper in search.results():            results.append({                "title": paper.title,                "authors": [a.name for a in paper.authors],                "summary": paper.summary,                "url": paper.pdf_url            })        return str(results)    async def _arun(self, query: str, max_results: int = 5) -> str:        return self._run(query, max_results)

# src/main.pyimport asynciofrom agent.planner import ResearchPlannerfrom agent.executor import ResearchExecutorfrom agent.memory import AgentMemoryfrom tools.arxiv import ArxivSearchToolfrom tools.scholar import ScholarSearchToolfrom tools.summarizer import PaperSummarizerasync def main():    # 初始化组件    planner = ResearchPlanner()    memory = AgentMemory()    tools = [        ArxivSearchTool(),        ScholarSearchTool(),        PaperSummarizer()    ]    executor = ResearchExecutor(tools, memory)    # 执行研究任务    topic = "大语言模型在代码生成领域的最新进展"    # 1. 规划    print("正在规划研究任务...")    plan = await planner.plan(topic)    print(f"研究计划: {plan}")    # 2. 执行    print("正在执行研究...")    results = await executor.execute(plan)    # 3. 生成报告    report = await generate_report(results, memory)    print(report)async def generate_report(results: dict, memory: AgentMemory) -> str:    """生成研究报告"""    from langchain_openai import ChatOpenAI    from langchain_core.prompts import ChatPromptTemplate    llm = ChatOpenAI(model="gpt-4")    prompt = ChatPromptTemplate.from_template("""    基于以下研究结果，生成一份研究报告：    研究结果：    {results}    对话历史：    {history}    请生成结构化的报告，包含：    1. 研究概述    2. 主要发现    3. 关键论文推荐    4. 未来研究方向    """)    chain = prompt | llm    return await chain.ainvoke({        "results": str(results),        "history": memory.get_context()    })if __name__ == "__main__":    asyncio.run(main())

┌─────────────────────────────────────────────────────────────────┐│                      Agent 框架能力雷达图                        │├─────────────────────────────────────────────────────────────────┤│                                                                 ││              生态完整性                                          ││                  ▲                                              ││      LangChain ●│● LangGraph                                   ││                 │                                               ││  易用性 ◄───────┼───────► 扩展性                               ││        CrewAI ● │ ● AutoGen                                    ││                 │                                               ││        AutoGPT ●│● MetaGPT                                     ││                  ▼                                              ││              自主性                                              ││                                                                 │└─────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────┐│                    Agent 框架选型指南                    │├─────────────────────────────────────────────────────────┤│                                                         ││  需要企业级稳定性和丰富生态？                            ││       │                                                 ││       ├── 是 ──▶ LangChain + LangGraph                 ││       │                                                 ││       └── 否 ──▶ 需要多 Agent 协作？                    ││                    │                                    ││                    ├── 是 ──▶ CrewAI 或 AutoGen        ││                    │                                    ││                    └── 否 ──▶ 需要完全自主？            ││                                 │                       ││                                 ├── 是 ──▶ AutoGPT     ││                                 │                       ││                                 └── 否 ──▶ 纯 Python   │└─────────────────────────────────────────────────────────┘

# LangGraph 示例：状态化工作流from langgraph.graph import StateGraph, ENDdef research_node(state):    # 研究节点    return {"research_done": True}def write_node(state):    # 写作节点    return {"article": "..."}workflow = StateGraph(State)workflow.add_node("research", research_node)workflow.add_node("write", write_node)workflow.add_edge("research", "write")workflow.add_edge("write", END)app = workflow.compile()

# CrewAI 示例：多 Agent 协作from crewai import Agent, Task, Crewresearcher = Agent(    role="研究员",    goal="收集相关信息",    backstory="资深研究专家",    allow_delegation=False)writer = Agent(    role="作家",    goal="撰写高质量文章",    backstory="专业内容创作者",    allow_delegation=True)task1 = Task(description="研究 AI 趋势", agent=researcher)task2 = Task(description="撰写报告", agent=writer)crew = Crew(agents=[researcher, writer], tasks=[task1, task2])result = crew.kickoff()

# AutoGen 示例：对话式多 Agentfrom autogen import AssistantAgent, UserProxyAgentassistant = AssistantAgent(    name="assistant",    llm_config={"model": "gpt-4"})user_proxy = UserProxyAgent(    name="user_proxy",    human_input_mode="NEVER",    code_execution_config={"work_dir": "coding"})user_proxy.initiate_chat(    assistant,    message="帮我写一个 Python 爬虫")

# 好的角色定义agent = Agent(    role="数据分析专家",    goal="从数据中提取有价值的洞察",    backstory="""    你是一位拥有10年经验的数据科学家，    擅长数据清洗、统计分析和可视化。    你总是以数据为依据，提供客观的分析结果。    """,    tools=[pandas_tool, matplotlib_tool])

# 工具应该是必要且有边界的tools = [    Tool(        name="search_web",        description="搜索互联网获取最新信息",        func=search_web,        # 设置边界        return_direct=False  # 不直接返回，允许 Agent 继续处理    )]

class SmartMemory:    def __init__(self):        self.working_memory = []  # 工作记忆（当前任务）        self.episodic_memory = [] # 情景记忆（历史交互）        self.semantic_memory = {} # 语义记忆（知识库）    def get_relevant_context(self, query: str) -> str:        """智能检索相关上下文"""        # 1. 从工作记忆获取最近信息        recent = self.working_memory[-5:]        # 2. 从语义记忆检索相关知识        relevant_knowledge = self.semantic_memory.search(query)        return self._merge_context(recent, relevant_knowledge)