AI Agent 是 LLM 驱动的自主系统能够感知环境、做出决策并执行动作。2.1 Agent 架构解析AI Agent 的核心架构由 Lilian WengOpenAI提出┌─────────────────────────────────────────────────────────┐ │ Agent 架构 │ ├─────────────────────────────────────────────────────────┤ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ │ Profile │ │ Memory │ │Planning │ │ Action │ │ │ │ 角色 │ │ 记忆 │ │ 规划 │ │ 行动 │ │ │ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │ │ │ │ │ │ │ │ └────────────┴────────────┴────────────┘ │ │ │ │ │ ┌─────▼─────┐ │ │ │ LLM │ │ │ │ (大脑) │ │ │ └───────────┘ │ └─────────────────────────────────────────────────────────┘核心组件详解组件功能实现方式Profile定义 Agent 角色与目标System PromptMemory短期/长期记忆存储Vector DB / RedisPlanning任务分解与规划CoT / ReAct / Plan-and-SolveAction工具调用与执行Function Calling2.2 Agent 类型对比2.2.1 ReAct Agent最经典的 Agent 模式结合推理Reasoning与行动ActingThought: 用户想知道今天的天气我需要先获取位置信息 Action: get_user_location Observation: 北京 Thought: 现在我知道了用户在北京需要查询北京天气 Action: get_weather(北京) Observation: 晴天25°C Answer: 北京今天天气晴朗气温25°CReAct 执行流程┌───────────────────────────────────────────────────┐ │ ReAct 循环 │ │ │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ │ Thought │───▶│ Action │───▶│Observation│ │ │ │ 思考 │ │ 行动 │ │ 观察 │ │ │ └─────────┘ └─────────┘ └─────────┘ │ │ ▲ │ │ │ └───────────────────────────────┘ │ │ (循环直到完成) │ └───────────────────────────────────────────────────┘2.2.2 Plan-and-Execute Agent适用于复杂任务先规划后执行# 规划阶段 plan 1. 收集用户需求文档 2. 分析技术可行性 3. 生成项目架构设计 4. 编写代码框架 5. 编写测试用例 ​ # 执行阶段 for step in plan.steps: result execute_step(step) if result.needs_revision: revise_plan(result.feedback)架构图用户输入 ──▶ 规划器 ──▶ 任务列表 ──▶ 执行器 ──▶ 结果 │ ▼ 反馈循环2.2.3 Multi-Agent System多 Agent 协作模拟团队工作┌──────────────────────────────────────────────────┐ │ Multi-Agent 系统 │ ├──────────────────────────────────────────────────┤ │ │ │ ┌────────┐ ┌────────┐ ┌────────┐ │ │ │ Leader │───▶│Analyst │ │ Coder │ │ │ │ Agent │ │ Agent │ │ Agent │ │ │ └────────┘ └────────┘ └────────┘ │ │ │ │ │ │ │ └─────────────┴─────────────┘ │ │ │ │ │ ┌───────▼───────┐ │ │ │ Shared Memory │ │ │ └───────────────┘ │ └──────────────────────────────────────────────────┘协作模式模式描述适用场景层级式Leader 分配任务给下属 Agent项目管理、软件开发对等式Agent 之间平等协作研究讨论、头脑风暴混合式根据任务动态调整协作方式复杂业务流程2.3 实战案例构建自动化研究 Agent场景描述构建一个研究 Agent能够自动搜索相关论文总结关键发现生成研究报告项目结构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.py import json from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate ​ class ResearchPlanner: def __init__(self, model: str gpt-4): self.llm ChatOpenAI(modelmodel) 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.py from typing import List, Dict from .memory import AgentMemory class 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.py from typing import List, Dict from dataclasses import dataclass, field from datetime import datetime dataclass class Message: role: str content: str timestamp: datetime field(default_factorydatetime.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(rolerole, contentcontent) 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.py from langchain.tools import BaseTool from pydantic import BaseModel, Field import arxiv class ArxivSearchInput(BaseModel): query: str Field(description搜索关键词) max_results: int Field(default5, 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( queryquery, max_resultsmax_results, sort_byarxiv.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.py import asyncio from agent.planner import ResearchPlanner from agent.executor import ResearchExecutor from agent.memory import AgentMemory from tools.arxiv import ArxivSearchTool from tools.scholar import ScholarSearchTool from tools.summarizer import PaperSummarizer async 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(modelgpt-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())2.4 Agent 框架对比与选型主流框架概览框架特点适用场景GitHub StarsLangChain生态最完善模块化设计企业级应用100kLangGraph状态化工作流可视化复杂任务流40kCrewAI多 Agent 协作角色驱动团队协作场景50kAutoGen (微软)对话式多 Agent代码生成编程辅助40kAutoGPT自主决策目标驱动个人助手170kMetaGPT模拟企业角色软件生成MVP 原型开发60k详细对比┌─────────────────────────────────────────────────────────────────┐ │ Agent 框架能力雷达图 │ ├─────────────────────────────────────────────────────────────────┤ │ │ │ 生态完整性 │ │ ▲ │ │ LangChain ●│● LangGraph │ │ │ │ │ 易用性 ◄───────┼───────► 扩展性 │ │ CrewAI ● │ ● AutoGen │ │ │ │ │ AutoGPT ●│● MetaGPT │ │ ▼ │ │ 自主性 │ │ │ └─────────────────────────────────────────────────────────────────┘选型决策树┌─────────────────────────────────────────────────────────┐ │ Agent 框架选型指南 │ ├─────────────────────────────────────────────────────────┤ │ │ │ 需要企业级稳定性和丰富生态 │ │ │ │ │ ├── 是 ──▶ LangChain LangGraph │ │ │ │ │ └── 否 ──▶ 需要多 Agent 协作 │ │ │ │ │ ├── 是 ──▶ CrewAI 或 AutoGen │ │ │ │ │ └── 否 ──▶ 需要完全自主 │ │ │ │ │ ├── 是 ──▶ AutoGPT │ │ │ │ │ └── 否 ──▶ 纯 Python │ └─────────────────────────────────────────────────────────┘框架特点详解LangChain LangGraph# LangGraph 示例状态化工作流 from langgraph.graph import StateGraph, END def 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# CrewAI 示例多 Agent 协作 from crewai import Agent, Task, Crew researcher Agent( role研究员, goal收集相关信息, backstory资深研究专家, allow_delegationFalse ) writer Agent( role作家, goal撰写高质量文章, backstory专业内容创作者, allow_delegationTrue ) task1 Task(description研究 AI 趋势, agentresearcher) task2 Task(description撰写报告, agentwriter) crew Crew(agents[researcher, writer], tasks[task1, task2]) result crew.kickoff()AutoGen# AutoGen 示例对话式多 Agent from autogen import AssistantAgent, UserProxyAgent assistant AssistantAgent( nameassistant, llm_config{model: gpt-4} ) user_proxy UserProxyAgent( nameuser_proxy, human_input_modeNEVER, code_execution_config{work_dir: coding} ) user_proxy.initiate_chat( assistant, message帮我写一个 Python 爬虫 )2.5 GitHub 项目推荐项目描述链接AutoGPT自主 AI Agent 先驱https://github.com/Significant-Gravitas/AutoGPTCrewAI多 Agent 协作框架https://github.com/joaomdmoura/crewAIAutoGen微软多 Agent 框架https://github.com/microsoft/autogenMetaGPT软件 Agent 团队https://github.com/geekan/MetaGPTAgentGPTWeb 端 Agent 部署https://github.com/reworkd/AgentGPTGPT Researcher研究 Agenthttps://github.com/assafelovic/gpt-researcherOpen Interpreter代码执行 Agenthttps://github.com/OpenInterpreter/open-interpreter2.6 最佳实践1. 清晰的角色定义# 好的角色定义 agent Agent( role数据分析专家, goal从数据中提取有价值的洞察, backstory 你是一位拥有10年经验的数据科学家 擅长数据清洗、统计分析和可视化。 你总是以数据为依据提供客观的分析结果。 , tools[pandas_tool, matplotlib_tool] )2. 合理的工具选择# 工具应该是必要且有边界的 tools [ Tool( namesearch_web, description搜索互联网获取最新信息, funcsearch_web, # 设置边界 return_directFalse # 不直接返回允许 Agent 继续处理 ) ]3. 记忆管理策略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)点击链接加入群聊【AI人工智能大模型交流QQ群】https://qm.qq.com/q/e9PYTfVW2k欢迎关注的我的公众号《码上未来》一起交流AI前沿技术!扫码二维码加我微信进群聊AI