Agent核心API详解

版本基线：本文基于 LangChain 1.x / LangGraph 1.x（2025-10 GA），示例模型 gpt-5-mini，更新于 2026-09。

概述

本章详细介绍Agent开发中的核心API，包括LLM引擎、工具注册、记忆管理、规划器、执行器等关键组件的API设计和使用方法。

1. LLM引擎API

LLM引擎是Agent的核心推理组件，负责理解用户输入、生成响应和决策。

1.1 基础LLM调用

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

# 创建LLM实例
llm = ChatOpenAI(
    model="gpt-5-mini",                # 模型名称
    temperature=0.7,                   # 温度参数（0-2），控制随机性
    max_tokens=2000,                   # 最大输出token数
    timeout=30,                        # 请求超时时间（秒）
    max_retries=2,                     # 最大重试次数
    api_key="your-api-key"            # API密钥
)

# 基础调用
response = llm.invoke([
    SystemMessage(content="你是一个专业的AI助手"),
    HumanMessage(content="请解释什么是Agent")
])

print(response.content)

1.2 流式输出

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model="gpt-5-mini", streaming=True)

# 流式调用
for chunk in llm.stream([HumanMessage(content="写一个关于AI的故事")]):
    print(chunk.content, end="", flush=True)

1.3 结构化输出

from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from typing import List

# 定义输出结构
class AgentDecision(BaseModel):
    """Agent决策结果"""
    reasoning: str = Field(description="推理过程")
    action: str = Field(description="要执行的动作")
    action_input: dict = Field(description="动作参数")
    confidence: float = Field(description="置信度 0-1")

# 创建支持结构化输出的LLM
llm = ChatOpenAI(model="gpt-5-mini")
structured_llm = llm.with_structured_output(AgentDecision)

# 调用
result = structured_llm.invoke([
    HumanMessage(content="用户问：今天天气怎么样？请决定下一步动作")
])

print(f"推理: {result.reasoning}")
print(f"动作: {result.action}")
print(f"参数: {result.action_input}")
print(f"置信度: {result.confidence}")

1.4 多模型切换

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI

class LLMManager:
    """LLM管理器 - 支持多模型切换"""
    
    def __init__(self):
        self.models = {
            "gpt-5-mini": ChatOpenAI(model="gpt-5-mini"),          # 日常任务
            "gpt-5.4": ChatOpenAI(model="gpt-5.4"),                # 旗舰，复杂推理
            "claude-sonnet-4-5": ChatAnthropic(model="claude-sonnet-4-5"),
            "gemini": ChatGoogleGenerativeAI(model="gemini-2.5-flash")
        }
        self.default_model = "gpt-5-mini"
    
    def get_model(self, model_name: str = None):
        """获取模型实例"""
        name = model_name or self.default_model
        return self.models.get(name)
    
    def invoke(self, messages, model_name: str = None):
        """调用指定模型"""
        model = self.get_model(model_name)
        return model.invoke(messages)

# 使用示例
manager = LLMManager()
response = manager.invoke(
    [HumanMessage(content="你好")],
    model_name="claude-sonnet-4-5"
)

2. 工具注册API

工具是Agent与外部世界交互的接口，包括搜索、计算、API调用等功能。

2.1 @tool装饰器

from langchain_core.tools import tool
from typing import Optional

# 使用@tool装饰器定义工具
@tool
def search_web(query: str, max_results: int = 5) -> str:
    """搜索互联网获取最新信息
    
    Args:
        query: 搜索关键词
        max_results: 最大返回结果数
    
    Returns:
        搜索结果的文本摘要
    """
    # 这里实现实际的搜索逻辑
    return f"搜索 '{query}' 的结果：找到 {max_results} 条相关信息"

@tool
def calculate(expression: str) -> str:
    """计算数学表达式
    
    Args:
        expression: 数学表达式，如 "2 + 3 * 4"
    
    Returns:
        计算结果
    """
    try:
        result = eval(expression)
        return f"计算结果: {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"

@tool
def read_file(file_path: str) -> str:
    """读取文件内容
    
    Args:
        file_path: 文件路径
    
    Returns:
        文件内容
    """
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            return f.read()
    except Exception as e:
        return f"读取文件失败: {str(e)}"

# 查看工具信息
print(search_web.name)           # search_web
print(search_web.description)    # 搜索互联网获取最新信息
print(search_web.args_schema.schema())  # 参数JSON Schema

2.2 StructuredTool

from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field
from typing import List, Optional

# 定义参数Schema
class SearchInput(BaseModel):
    """搜索参数"""
    query: str = Field(description="搜索关键词")
    num_results: int = Field(default=5, description="返回结果数量")
    language: str = Field(default="zh", description="搜索语言")

class SearchResult(BaseModel):
    """搜索结果"""
    title: str = Field(description="标题")
    url: str = Field(description="链接")
    snippet: str = Field(description="摘要")

# 实现工具函数
def search_function(query: str, num_results: int = 5, language: str = "zh") -> List[SearchResult]:
    """执行搜索"""
    # 实际搜索逻辑
    return [
        SearchResult(title=f"结果{i}", url=f"https://example.com/{i}", snippet=f"摘要{i}")
        for i in range(num_results)
    ]

# 创建StructuredTool
search_tool = StructuredTool.from_function(
    func=search_function,
    name="advanced_search",
    description="高级搜索工具，支持多语言和结果数量配置",
    args_schema=SearchInput,
    return_direct=True  # 直接返回结果给用户
)

# 使用工具
result = search_tool.invoke({"query": "Python Agent", "num_results": 3})
print(result)

2.3 工具注册表

from langchain_core.tools import BaseTool, tool
from typing import Dict, List, Optional, Any
from pydantic import BaseModel, Field

class ToolRegistry:
    """工具注册表 - 管理所有可用工具"""
    
    def __init__(self):
        self._tools: Dict[str, BaseTool] = {}
        self._categories: Dict[str, List[str]] = {}
    
    def register(self, tool: BaseTool, category: str = "general"):
        """注册工具
        
        Args:
            tool: 工具实例
            category: 工具类别
        """
        self._tools[tool.name] = tool
        
        if category not in self._categories:
            self._categories[category] = []
        self._categories[category].append(tool.name)
    
    def get_tool(self, name: str) -> Optional[BaseTool]:
        """获取工具"""
        return self._tools.get(name)
    
    def get_tools(self, category: str = None) -> List[BaseTool]:
        """获取工具列表"""
        if category:
            tool_names = self._categories.get(category, [])
            return [self._tools[name] for name in tool_names if name in self._tools]
        return list(self._tools.values())
    
    def get_tool_names(self) -> List[str]:
        """获取所有工具名称"""
        return list(self._tools.keys())
    
    def remove_tool(self, name: str):
        """移除工具"""
        if name in self._tools:
            del self._tools[name]
            for category, tools in self._categories.items():
                if name in tools:
                    tools.remove(name)

# 使用示例
registry = ToolRegistry()

# 注册工具
registry.register(search_web, category="search")
registry.register(calculate, category="utility")
registry.register(read_file, category="file")

# 获取特定类别的工具
search_tools = registry.get_tools(category="search")
print(f"搜索工具: {[t.name for t in search_tools]}")

# 获取所有工具
all_tools = registry.get_tools()
print(f"所有工具: {[t.name for t in all_tools]}")

2.4 动态工具加载

import importlib
from langchain_core.tools import BaseTool
from typing import Dict, Any

class DynamicToolLoader:
    """动态工具加载器"""
    
    def __init__(self):
        self.loaded_tools: Dict[str, BaseTool] = {}
    
    def load_from_module(self, module_path: str, tool_names: List[str] = None):
        """从模块加载工具
        
        Args:
            module_path: 模块路径，如 "my_tools.search"
            tool_names: 要加载的工具名称列表，None表示加载所有
        """
        try:
            module = importlib.import_module(module_path)
            
            # 获取模块中所有工具
            for attr_name in dir(module):
                attr = getattr(module, attr_name)
                if isinstance(attr, BaseTool):
                    if tool_names is None or attr_name in tool_names:
                        self.loaded_tools[attr_name] = attr
                        print(f"加载工具: {attr_name}")
        except Exception as e:
            print(f"加载模块失败: {e}")
    
    def load_from_config(self, config: Dict[str, Any]):
        """从配置加载工具
        
        Args:
            config: 工具配置字典
            {
                "search": {
                    "module": "tools.search",
                    "class": "SearchTool",
                    "params": {"api_key": "xxx"}
                }
            }
        """
        for tool_name, tool_config in config.items():
            try:
                module = importlib.import_module(tool_config["module"])
                tool_class = getattr(module, tool_config["class"])
                tool_instance = tool_class(**tool_config.get("params", {}))
                self.loaded_tools[tool_name] = tool_instance
            except Exception as e:
                print(f"加载工具 {tool_name} 失败: {e}")
    
    def get_tool(self, name: str) -> Optional[BaseTool]:
        """获取工具"""
        return self.loaded_tools.get(name)

3. 记忆管理API

记忆系统让Agent能够记住对话历史和重要信息。

3.1 对话记忆

LangChain 1.x 移除了 langchain.memory.*，对话记忆统一交给 LangGraph checkpointer 管理：给 Agent 挂载一个 checkpointer，再用 thread_id 区分会话，每轮对话状态会自动保存和恢复。

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver

# 1. 创建 checkpointer（对话记忆的载体）
checkpointer = InMemorySaver()  # 内存版；生产可用 SqliteSaver / PostgresSaver

# 2. 创建 Agent 时挂载 checkpointer
agent = create_agent(
    ChatOpenAI(model="gpt-5-mini"),
    tools=[],
    system_prompt="你是一个贴心的助手。",
    checkpointer=checkpointer,
)

# 3. 用 thread_id 区分不同会话，记忆自动持久化
config = {"configurable": {"thread_id": "user-zhangsan"}}

agent.invoke(
    {"messages": [{"role": "user", "content": "你好，我叫张三"}]},
    config,
)
agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么名字？"}]},
    config,
)
# → Agent 能答出"张三"，因为同一 thread_id 的消息历史被自动恢复

# 查看该会话的完整消息历史
state = agent.get_state(config)
for msg in state.values["messages"]:
    print(type(msg).__name__, ":", msg.content)

旧写法对照（LangChain 0.x）

旧版通过 langchain.memory 的各类 Memory 类手动管理对话历史，这些类在 1.x 中已移除（兼容实现迁入 langchain-classic 包）：

# ❌ 旧写法（LangChain 0.x，已移除）
from langchain.memory import (
    ConversationBufferMemory,
    ConversationSummaryMemory,
    ConversationBufferWindowMemory
)
memory = ConversationBufferMemory(return_messages=True, memory_key="history")
memory.save_context({"input": "..."}, {"output": "..."})
variables = memory.load_memory_variables({})

新方案把"存记忆"变成"存状态"：checkpointer 负责持久化，thread_id 负责隔离会话；窗口截断、历史摘要等策略可用自定义 middleware 实现。

3.2 向量记忆

旧版 VectorStoreRetrieverMemory 已随 langchain.memory 移除。现代做法是直接使用向量库集成包存取记忆（Chroma 等已拆分为独立包），再把检索能力包装成工具交给 Agent：

# pip install langchain-chroma
from langchain_chroma import Chroma          # 1.x 起为独立集成包
from langchain_openai import OpenAIEmbeddings

class VectorMemory:
    """向量记忆 - 基于语义检索历史信息"""

    def __init__(self, collection_name: str = "agent_memory"):
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        self.vectorstore = Chroma(
            collection_name=collection_name,
            embedding_function=self.embeddings,
        )

    def save(self, text: str, metadata: dict | None = None):
        """把一条记忆写入向量库"""
        self.vectorstore.add_texts(texts=[text], metadatas=[metadata or {}])

    def recall(self, query: str, k: int = 5) -> list[str]:
        """根据语义相似度召回相关记忆"""
        docs = self.vectorstore.similarity_search(query, k=k)
        return [doc.page_content for doc in docs]

# 使用示例
vector_memory = VectorMemory()

# 保存记忆
vector_memory.save("用户喜欢吃苹果", {"type": "偏好"})
vector_memory.save("用户住在北京", {"type": "信息"})

# 召回相关记忆
relevant_memories = vector_memory.recall("我喜欢什么水果？")
print(f"相关记忆: {relevant_memories}")

把 recall 用 @tool 包装后挂到 create_agent 上，Agent 就能在需要时主动"回忆"相关上下文。

向量库去哪了

langchain_community.vectorstores 中的主流向量库已陆续拆分为独立集成包：Chroma → langchain-chroma、MongoDB → langchain-mongodb 等；FAISS、PGVector 等仍在 langchain_community 中维护。

3.3 长期记忆

import json
import sqlite3
from datetime import datetime
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, asdict

@dataclass
class MemoryEntry:
    """记忆条目"""
    id: str
    timestamp: str
    category: str
    content: str
    metadata: Dict[str, Any]
    importance: float  # 0-1

class LongTermMemory:
    """长期记忆 - 基于SQLite的持久化记忆"""
    
    def __init__(self, db_path: str = "agent_memory.db"):
        self.db_path = db_path
        self._init_db()
    
    def _init_db(self):
        """初始化数据库"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS memories (
                id TEXT PRIMARY KEY,
                timestamp TEXT,
                category TEXT,
                content TEXT,
                metadata TEXT,
                importance REAL
            )
        """)
        conn.commit()
        conn.close()
    
    def save(self, category: str, content: str, 
             metadata: Dict = None, importance: float = 0.5) -> str:
        """保存记忆"""
        memory_id = f"{category}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
        entry = MemoryEntry(
            id=memory_id,
            timestamp=datetime.now().isoformat(),
            category=category,
            content=content,
            metadata=metadata or {},
            importance=importance
        )
        
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute(
            "INSERT INTO memories VALUES (?, ?, ?, ?, ?, ?)",
            (entry.id, entry.timestamp, entry.category, 
             entry.content, json.dumps(entry.metadata), entry.importance)
        )
        conn.commit()
        conn.close()
        
        return memory_id
    
    def recall(self, category: str = None, limit: int = 10) -> List[MemoryEntry]:
        """召回记忆"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        if category:
            cursor.execute(
                "SELECT * FROM memories WHERE category = ? ORDER BY importance DESC, timestamp DESC LIMIT ?",
                (category, limit)
            )
        else:
            cursor.execute(
                "SELECT * FROM memories ORDER BY importance DESC, timestamp DESC LIMIT ?",
                (limit,)
            )
        
        rows = cursor.fetchall()
        conn.close()
        
        return [
            MemoryEntry(
                id=row[0], timestamp=row[1], category=row[2],
                content=row[3], metadata=json.loads(row[4]), importance=row[5]
            )
            for row in rows
        ]
    
    def search(self, keyword: str, limit: int = 10) -> List[MemoryEntry]:
        """搜索记忆"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute(
            "SELECT * FROM memories WHERE content LIKE ? ORDER BY importance DESC LIMIT ?",
            (f"%{keyword}%", limit)
        )
        rows = cursor.fetchall()
        conn.close()
        
        return [
            MemoryEntry(
                id=row[0], timestamp=row[1], category=row[2],
                content=row[3], metadata=json.loads(row[4]), importance=row[5]
            )
            for row in rows
        ]

# 使用示例
ltm = LongTermMemory()

# 保存重要信息
ltm.save("user_info", "用户喜欢Python编程", {"source": "conversation"}, importance=0.8)
ltm.save("task", "完成了RAG系统开发", {"project": "ai-assistant"}, importance=0.9)

# 召回记忆
user_memories = ltm.recall(category="user_info")
print(f"用户相关记忆: {[m.content for m in user_memories]}")

4. 规划器API

规划器负责将复杂任务分解为可执行的步骤。

4.1 任务分解

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from typing import List

class TaskStep(BaseModel):
    """任务步骤"""
    step_number: int = Field(description="步骤编号")
    description: str = Field(description="步骤描述")
    required_tools: List[str] = Field(description="需要的工具")
    dependencies: List[int] = Field(default=[], description="依赖的步骤编号")

class TaskPlan(BaseModel):
    """任务计划"""
    goal: str = Field(description="任务目标")
    steps: List[TaskStep] = Field(description="执行步骤")
    estimated_time: str = Field(description="预估时间")

class TaskPlanner:
    """任务规划器"""
    
    def __init__(self, llm=None):
        self.llm = llm or ChatOpenAI(model="gpt-5-mini")
        self.structured_llm = self.llm.with_structured_output(TaskPlan)
    
    def create_plan(self, task: str, available_tools: List[str]) -> TaskPlan:
        """创建任务执行计划
        
        Args:
            task: 任务描述
            available_tools: 可用工具列表
        
        Returns:
            任务计划
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一个任务规划专家。根据用户的任务描述，创建详细的执行计划。
            
可用工具: {tools}

请将任务分解为清晰的步骤，每一步都要指定需要使用的工具。"""),
            ("human", "请为以下任务创建执行计划: {task}")
        ])
        
        chain = prompt | self.structured_llm
        
        plan = chain.invoke({
            "task": task,
            "tools": ", ".join(available_tools)
        })
        
        return plan
    
    def replan(self, original_plan: TaskPlan, 
               completed_steps: List[int], 
               error_info: str = None) -> TaskPlan:
        """重新规划
        
        Args:
            original_plan: 原始计划
            completed_steps: 已完成的步骤
            error_info: 错误信息（如果有）
        
        Returns:
            新的任务计划
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一个任务规划专家。根据执行情况重新规划任务。

原始计划: {original_plan}
已完成步骤: {completed_steps}
错误信息: {error_info}

请创建新的执行计划，调整未完成的步骤。"""),
            ("human", "请重新规划任务")
        ])
        
        chain = prompt | self.structured_llm
        
        new_plan = chain.invoke({
            "original_plan": original_plan.model_dump_json(),
            "completed_steps": str(completed_steps),
            "error_info": error_info or "无"
        })
        
        return new_plan

# 使用示例
planner = TaskPlanner()

# 创建计划
plan = planner.create_plan(
    task="帮我搜索最新的AI新闻，总结要点，并发送邮件给团队",
    available_tools=["search_web", "summarize_text", "send_email"]
)

print(f"任务目标: {plan.goal}")
print(f"执行步骤:")
for step in plan.steps:
    print(f"  {step.step_number}. {step.description} (工具: {step.required_tools})")

4.2 ReAct规划器

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import HumanMessage, AIMessage
from typing import List, Dict, Any, Tuple

class ReActPlanner:
    """ReAct规划器 - 推理与行动交替执行"""
    
    def __init__(self, llm=None):
        self.llm = llm or ChatOpenAI(model="gpt-5-mini")
    
    def think(self, task: str, context: str, 
              available_tools: List[str]) -> Tuple[str, str, Dict]:
        """思考下一步行动
        
        Args:
            task: 任务描述
            context: 当前上下文（历史动作和观察结果）
            available_tools: 可用工具
        
        Returns:
            (思考过程, 动作名称, 动作参数)
        """
        prompt = f"""你是一个智能Agent。根据任务和当前情况，决定下一步行动。

任务: {task}

可用工具: {', '.join(available_tools)}

历史动作和观察:
{context}

请按以下格式回答:
Thought: 分析当前情况，思考下一步
Action: 工具名称
Action Input: 工具参数（JSON格式）

如果任务完成，回答:
Thought: 任务已完成
Final Answer: 最终答案"""

        response = self.llm.invoke([HumanMessage(content=prompt)])
        
        # 解析响应
        content = response.content
        thought = ""
        action = ""
        action_input = {}
        
        for line in content.split("\n"):
            if line.startswith("Thought:"):
                thought = line[8:].strip()
            elif line.startswith("Action:"):
                action = line[7:].strip()
            elif line.startswith("Action Input:"):
                try:
                    action_input = json.loads(line[13:].strip())
                except:
                    action_input = {"input": line[13:].strip()}
            elif line.startswith("Final Answer:"):
                return "任务完成", "finish", {"answer": line[13:].strip()}
        
        return thought, action, action_input

# 使用示例
react_planner = ReActPlanner()

task = "搜索今天的天气并告诉用户"
context = ""
tools = ["search_weather", "send_message"]

# 模拟ReAct循环
for i in range(5):  # 最多5轮
    thought, action, action_input = react_planner.think(task, context, tools)
    
    print(f"\n--- 第{i+1}轮 ---")
    print(f"思考: {thought}")
    print(f"动作: {action}")
    print(f"参数: {action_input}")
    
    if action == "finish":
        print(f"最终答案: {action_input.get('answer')}")
        break
    
    # 模拟执行动作
    observation = f"执行了 {action}，参数为 {action_input}"
    context += f"\nThought: {thought}\nAction: {action}\nAction Input: {action_input}\nObservation: {observation}\n"

5. 执行器API

执行器负责实际执行工具调用和处理结果。

5.1 基础执行器

from typing import List, Dict, Any

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

class AgentManager:
    """Agent管理器（LangChain 1.x：create_agent 取代 AgentExecutor）"""

    def __init__(self, llm=None, tools=None):
        self.llm = llm or ChatOpenAI(model="gpt-5-mini")
        self.tools = tools or []
        self.agent = None

    def create(self, system_prompt: str):
        """创建 Agent

        Args:
            system_prompt: 系统提示词

        Returns:
            可调用的 Agent（LangGraph 图）
        """
        self.agent = create_agent(
            self.llm,
            tools=self.tools,
            system_prompt=system_prompt,
        )
        return self.agent

    def execute(self, input_text: str) -> Dict[str, Any]:
        """执行任务

        Args:
            input_text: 用户输入

        Returns:
            执行结果
        """
        if not self.agent:
            raise ValueError("请先调用 create 创建 Agent")

        result = self.agent.invoke({
            "messages": [{"role": "user", "content": input_text}]
        })

        return {
            "output": result["messages"][-1].content,
            "messages": result["messages"],  # 完整消息轨迹（含工具调用过程）
        }

# 使用示例
@tool
def search(query: str) -> str:
    """搜索信息"""
    return f"搜索结果: {query}"

@tool
def calculate(expression: str) -> str:
    """计算表达式"""
    return str(eval(expression))  # ⚠️ 教学演示，生产环境请改用白名单解析/沙箱

# 创建执行器
manager = AgentManager(
    tools=[search, calculate]
)

manager.create(
    system_prompt="你是一个有用的AI助手，可以使用搜索和计算工具。"
)

# 执行任务
result = manager.execute("搜索今天的新闻，然后计算3+5等于多少")
print(f"输出: {result['output']}")
print(f"消息数: {len(result['messages'])}")

旧写法对照（AgentExecutor 常用参数）

旧版 AgentExecutor 的常用控制参数在 LangChain 1.x 中的对应做法：

旧参数	新做法
max_iterations=10	LangGraph 的 config={"recursion_limit": 10}
verbose=True	用 graph.stream() 逐步观察，或接入 LangSmith
handle_parsing_errors=True	在 middleware 中捕获并兜底处理
return_intermediate_steps=True	直接读取结果中的完整 messages 轨迹

5.2 错误处理和重试

from typing import Any, Dict
import time

class RobustAgent:
    """健壮的Agent执行 - 支持错误处理和重试"""

    def __init__(self, agent, max_retries: int = 3):
        self.agent = agent  # create_agent 返回的 Agent（LangGraph 图）
        self.max_retries = max_retries

    def execute_with_retry(self, input_text: str) -> Dict[str, Any]:
        """带重试的执行

        Args:
            input_text: 用户输入

        Returns:
            执行结果
        """
        last_error = None

        for attempt in range(self.max_retries):
            try:
                result = self.agent.invoke({
                    "messages": [{"role": "user", "content": input_text}]
                })
                return {
                    "success": True,
                    "result": result["messages"][-1].content,
                    "attempts": attempt + 1
                }
            except Exception as e:
                last_error = e
                print(f"第 {attempt + 1} 次执行失败: {str(e)}")

                if attempt < self.max_retries - 1:
                    # 指数退避
                    wait_time = 2 ** attempt
                    print(f"等待 {wait_time} 秒后重试...")
                    time.sleep(wait_time)

        return {
            "success": False,
            "error": str(last_error),
            "attempts": self.max_retries
        }

# 使用示例（agent 来自上一个示例中 manager.create(...) 创建的 Agent）
robust_agent = RobustAgent(manager.agent, max_retries=3)
result = robust_agent.execute_with_retry("查询天气")

6. 状态管理API

状态管理让Agent能够维护和更新执行过程中的状态信息。

6.1 状态定义

from typing import TypedDict, Annotated, List, Dict, Any
from operator import add

class AgentState(TypedDict):
    """Agent状态定义"""
    # 用户输入
    input: str
    
    # 对话历史（使用add操作符合并）
    messages: Annotated[List[Dict], add]
    
    # 当前任务
    current_task: str
    
    # 任务计划
    task_plan: Dict[str, Any]
    
    # 已完成的步骤
    completed_steps: Annotated[List[str], add]
    
    # 工具执行结果
    tool_results: Annotated[List[Dict], add]
    
    # 最终输出
    output: str
    
    # 错误信息
    error: str
    
    # 元数据
    metadata: Dict[str, Any]

6.2 状态管理器

from typing import Dict, Any, Optional
import json
from datetime import datetime

class StateManager:
    """状态管理器"""
    
    def __init__(self):
        self.state: Dict[str, Any] = {}
        self.history: List[Dict[str, Any]] = []
    
    def initialize(self, initial_state: Dict[str, Any]):
        """初始化状态"""
        self.state = initial_state.copy()
        self._save_snapshot("initialize")
    
    def get(self, key: str, default=None) -> Any:
        """获取状态值"""
        return self.state.get(key, default)
    
    def set(self, key: str, value: Any):
        """设置状态值"""
        old_value = self.state.get(key)
        self.state[key] = value
        self._save_snapshot(f"set_{key}")
    
    def update(self, updates: Dict[str, Any]):
        """批量更新状态"""
        self.state.update(updates)
        self._save_snapshot("batch_update")
    
    def append(self, key: str, value: Any):
        """追加到列表状态"""
        if key not in self.state:
            self.state[key] = []
        self.state[key].append(value)
        self._save_snapshot(f"append_{key}")
    
    def get_snapshot(self) -> Dict[str, Any]:
        """获取当前状态快照"""
        return self.state.copy()
    
    def _save_snapshot(self, action: str):
        """保存状态快照"""
        self.history.append({
            "timestamp": datetime.now().isoformat(),
            "action": action,
            "state": self.state.copy()
        })
    
    def get_history(self) -> List[Dict[str, Any]]:
        """获取状态历史"""
        return self.history
    
    def rollback(self, steps: int = 1):
        """回滚到之前的状态"""
        if len(self.history) > steps:
            self.history = self.history[:-steps]
            self.state = self.history[-1]["state"].copy()

# 使用示例
state_manager = StateManager()

# 初始化状态
state_manager.initialize({
    "input": "搜索天气",
    "messages": [],
    "completed_steps": [],
    "output": ""
})

# 更新状态
state_manager.set("current_task", "天气查询")
state_manager.append("completed_steps", "step1_search")
state_manager.update({
    "tool_results": [{"tool": "search", "result": "晴天"}],
    "output": "今天天气晴朗"
})

# 获取状态
print(f"当前任务: {state_manager.get('current_task')}")
print(f"已完成步骤: {state_manager.get('completed_steps')}")
print(f"状态历史: {len(state_manager.get_history())} 条记录")

总结

本章介绍了Agent开发的六大核心API：

API	功能	关键类/函数
LLM引擎	模型调用、流式输出、结构化输出	ChatOpenAI, with_structured_output
工具注册	定义、注册、管理工具	@tool, StructuredTool, ToolRegistry
记忆管理	对话记忆、向量记忆、长期记忆	InMemorySaver(checkpointer), VectorMemory
规划器	任务分解、ReAct规划	TaskPlanner, ReActPlanner
执行器	执行循环、错误处理、重试	create_agent, RobustAgent
状态管理	状态定义、更新、历史记录	AgentState, StateManager

下一步学习

• Agent核心技术详解 - 深入理解ReAct、CoT等技术
• LangChain框架 - 学习LangChain的Agent实现
• LangGraph工作流 - 学习基于图的Agent工作流