graph TB
A[AI Agent] --> B[感知 Perception]
A --> C[推理 Reasoning]
A --> D[行动 Action]
A --> E[记忆 Memory]
B --> B1[用户输入]
B --> B2[环境观察]
B --> B3[工具返回值]
C --> C1[任务分解]
C --> C2[计划生成]
C --> C3[决策选择]
D --> D1[调用工具]
D --> D2[生成回复]
D --> D3[修改环境]
E --> E1[短期记忆<br/>对话上下文]
E --> E2[工作记忆<br/>当前任务状态]
E --> E3[长期记忆<br/>知识库 / 经验]
style A fill:#2c3e50,color:#fff
style C fill:#e74c3c,color:#fff
style D fill:#3498db,color:#fff
style E fill:#2ecc71,color:#fff
from langchain.agents import create_react_agent, AgentExecutor from langchain_core.prompts import PromptTemplate
REACT_PROMPT = PromptTemplate.from_template("""Answer the following questions as best you can. You have access to the following tools:
{tools}
Use the following format:
Question: the input question you must answer Thought: you should always think about what to do Action: the action to take, should be one of [{tool_names}] Action Input: the input to the action Observation: the result of the action ... (this Thought/Action/Action Input/Observation can repeat N times) Thought: I now know the final answer Final Answer: the final answer to the original input question
result = executor.invoke({"input": "北京和上海哪个城市今天更热?"})
规划策略
Plan-and-Execute
先制定完整计划，再逐步执行——适合复杂的多步任务：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
from langchain.agents import create_plan_and_execute_agent from langchain_experimental.plan_and_execute import PlanAndExecute, load_agent_executor, load_chat_planner
# Planner: generates step-by-step plan planner = load_chat_planner(llm)
# Executor: executes each step executor_agent = load_agent_executor(llm, tools, verbose=True)
result = agent.invoke({ "input": "帮我调研目前主流的向量数据库，对比它们的性能，然后写一份推荐报告" })
graph TD
A[复杂任务] --> B[Planner<br/>LLM 生成计划]
B --> C["Step 1: 列出主流向量数据库"]
C --> D["Step 2: 搜索各数据库性能数据"]
D --> E["Step 3: 制作对比表格"]
E --> F["Step 4: 撰写推荐报告"]
C --> G[Executor<br/>执行并观察]
D --> G
E --> G
F --> G
G --> H{需要重新规划?}
H -->|是| B
H -->|否| I[最终结果]
style B fill:#e74c3c,color:#fff
style G fill:#3498db,color:#fff
style I fill:#2ecc71,color:#fff
defplanner_node(state: AgentState) -> AgentState: """Generate or revise the plan.""" messages = state["messages"] plan_prompt = f"""Based on the task and current progress, create a plan.
plan = llm.invoke(plan_prompt).content steps = [s.strip() for s in plan.split("\n") if s.strip()] return {"plan": steps, "current_step": 0}
defexecutor_node(state: AgentState) -> AgentState: """Execute the current step.""" step = state["plan"][state["current_step"]] result = agent_executor.invoke({"input": step}) return { "results": {**state.get("results", {}), step: result["output"]}, "current_step": state["current_step"] + 1, }
defshould_continue(state: AgentState) -> str: """Decide whether to continue, replan, or finish.""" if state["current_step"] >= len(state["plan"]): return"finish" if state.get("should_replan"): return"replan" return"execute"
graph TB
A[Agent Memory Architecture] --> B[Sensory Memory<br/>当前输入]
A --> C[Short-term Memory<br/>对话窗口]
A --> D[Long-term Memory<br/>持久存储]
C --> C1[最近 N 轮对话]
C --> C2[当前任务上下文]
D --> D1[Episodic<br/>历史事件记忆]
D --> D2[Semantic<br/>知识库 / 向量存储]
D --> D3[Procedural<br/>技能 / 工具使用经验]
style B fill:#ffd700,color:#000
style C fill:#3498db,color:#fff
style D fill:#2ecc71,color:#fff
memories = [] if episodic: memories.append("相关历史记忆:\n" + "\n".join([d.page_content for d in episodic])) if semantic: memories.append("相关知识:\n" + "\n".join([d.page_content for d in semantic]))
return"\n\n".join(memories)
deflearn(self, knowledge: str, metadata: dict = None): """Store new knowledge in semantic memory.""" self.semantic_memory.add_texts( texts=[knowledge], metadatas=[metadata or {}], )
多 Agent 系统
Supervisor 模式
graph TB
A[User Query] --> B[Supervisor Agent]
B --> C{任务分配}
C --> D[Research Agent<br/>信息搜集]
C --> E[Code Agent<br/>代码编写]
C --> F[Review Agent<br/>质量审查]
D --> G[汇报结果]
E --> G
F --> G
G --> B
B --> H[综合回答]
style B fill:#e74c3c,color:#fff
style D fill:#3498db,color:#fff
style E fill:#f39c12,color:#000
style F fill:#2ecc71,color:#fff