跳到主要内容

AI Agent 知识速查表

本页面汇总了 AI Agent 开发中最常用的概念、API 和最佳实践，方便快速查阅。

核心概念​

概念说明
Agent能够感知环境、做出决策并采取行动的智能系统
LLM大语言模型，Agent 的"大脑"，负责推理和决策
Tool工具，Agent 与外部世界交互的桥梁
Memory记忆系统，存储对话历史和知识信息
Planning规划模块，将复杂任务分解为子任务
Task Decomposition任务分解，将大目标拆分为可执行的小步骤
ReAct推理与行动交替进行的执行模式
Function CallingLLM 调用外部函数的机制
LangGraph用于构建有状态 Agent 的图框架
Middleware中间件，用于扩展 Agent 能力
CoT (Chain of Thought)思维链，逐步推理的提示技术
ToT (Tree of Thoughts)思维树，探索多条推理路径的方法

Agent 公式​

经典公式​

AI Agent = LLM + Planning + Memory + Tools

LangChain 1.0 架构​

Agent = Model + Tools + System Prompt + State (+ Middleware)

LangChain 1.0 API（推荐）​

创建 Agent​

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

@tool
def my_tool(param: str) -> str:
    """工具描述（LLM 会读取这个描述）"""
    return f"结果：{param}"

llm = ChatOpenAI(model="gpt-4", temperature=0)

agent = create_agent(
    model=llm,
    tools=[my_tool],
    system_prompt="你是一个智能助手。"
)

# 调用 Agent
result = agent.invoke({
    "messages": [{"role": "user", "content": "用户输入"}]
})

print(result["messages"][-1].content)

流式输出​

for chunk in agent.stream({
    "messages": [{"role": "user", "content": "用户输入"}]
}):
    print(chunk)

带状态的 Agent​

from langchain.agents import AgentState
from typing import TypedDict

class MyState(AgentState):
    user_id: str
    preferences: dict

agent = create_agent(
    model=llm,
    tools=[...],
    state_schema=MyState
)

中间件​

from langchain.agents.middleware import AgentMiddleware

class LoggingMiddleware(AgentMiddleware):
    def before_model(self, state, context):
        """模型调用前执行"""
        print(f"调用模型")
        return {}  # 返回空字典表示不更新状态
    
    def after_model(self, state, context):
        """模型调用后执行"""
        print(f"模型响应")
        return {}
    
    def wrap_tool_call(self, request, handler):
        """包装工具调用"""
        try:
            result = yield handler(request)
            return result
        except Exception as e:
            print(f"工具错误：{e}")
            return f"工具执行出错，请尝试其他方法。"

agent = create_agent(
    model=llm,
    tools=[...],
    middleware=[LoggingMiddleware()]
)

结构化输出​

from pydantic import BaseModel, Field

class Answer(BaseModel):
    summary: str = Field(description="摘要")
    key_points: list[str] = Field(description="关键点")

agent = create_agent(
    model=llm,
    tools=[...],
    response_format=Answer
)

result = agent.invoke({...})
# result 将是结构化的 Answer 对象

工具定义​

基本工具​

from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """搜索互联网获取信息"""
    return f"搜索结果：{query}"

@tool
def calculator(expression: str) -> str:
    """计算数学表达式"""
    try:
        return str(eval(expression))
    except:
        return "计算错误"

带参数验证的工具​

from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field, validator

class WeatherInput(BaseModel):
    city: str = Field(description="城市名称")
    unit: str = Field(default="celsius", description="温度单位")
    
    @validator('city')
    def validate_city(cls, v):
        if len(v) < 2:
            raise ValueError('城市名称太短')
        return v

def get_weather(city: str, unit: str = "celsius") -> str:
    return f"{city}：25°{unit[0].upper()}"

weather_tool = StructuredTool(
    name="get_weather",
    description="获取城市天气",
    func=get_weather,
    args_schema=WeatherInput
)

记忆类型​

类型说明适用场景
ConversationBufferMemory保存所有对话短对话
ConversationBufferWindowMemory保留最近 N 轮长对话
ConversationTokenBufferMemory按 token 限制控制上下文长度
ConversationSummaryMemory自动总结超长对话
VectorStoreRetrieverMemory向量检索长期记忆

LangChain 1.0 记忆​

# Agent 自动维护对话历史
agent = create_agent(model=llm, tools=[...])

# 第一次调用
agent.invoke({"messages": [{"role": "user", "content": "我叫张三"}]})

# 第二次调用（Agent 记得之前的对话）
agent.invoke({"messages": [{"role": "user", "content": "我叫什么名字？"}]})

OpenAI Function Calling​

工具定义格式​

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取城市天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称"
                    }
                },
                "required": ["city"]
            }
        }
    }
]

直接调用​

from openai import OpenAI
import json

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "北京天气"}],
    tools=tools,
    tool_choice="auto"
)

if response.choices[0].message.tool_calls:
    tool_call = response.choices[0].message.tool_calls[0]
    function_name = tool_call.function.name
    arguments = json.loads(tool_call.function.arguments)
    # 执行函数...

LangGraph 状态图​

基本结构​

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
import operator

class State(TypedDict):
    messages: Annotated[list, operator.add]

def agent_node(state: State) -> dict:
    return {"messages": [response]}

workflow = StateGraph(State)
workflow.add_node("agent", agent_node)
workflow.set_entry_point("agent")
workflow.add_edge("agent", END)

app = workflow.compile()

条件路由​

def should_continue(state: State) -> str:
    if state["messages"][-1].tool_calls:
        return "tools"
    return "end"

workflow.add_conditional_edges(
    "agent",
    should_continue,
    {"tools": "tools", "end": END}
)

评估与测试​

LangSmith 配置​

import os

os.environ["LANGSMITH_API_KEY"] = "your-key"
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "my-project"

运行评估​

from langchain.smith import RunEvalConfig, run_on_dataset
from langsmith import Client

client = Client()

eval_config = RunEvalConfig(
    evaluators=["exact_match", "contains"],
    eval_llm=ChatOpenAI(model="gpt-4")
)

results = run_on_dataset(
    client=client,
    dataset_name="my-dataset",
    llm_or_chain_factory=lambda: agent,
    evaluation=eval_config
)

部署​

LangServe​

from langserve import add_routes
from fastapi import FastAPI
import uvicorn

app = FastAPI()
add_routes(app, agent, path="/agent")

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Docker​

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

常用提示词模板​

系统提示词​

SYSTEM_PROMPT = """
你是一个专业的助手，具备以下能力：
1. 信息搜索和分析
2. 任务规划和执行
3. 结果总结和报告

请根据用户需求，合理使用工具完成任务。
如果遇到问题，请说明原因并提供替代方案。
"""

ReAct 提示词​

问题：用户的问题
思考：你应该做什么
行动：工具名称
行动输入：工具参数
观察：工具返回结果
...（重复思考/行动/观察）
思考：我现在知道最终答案
最终答案：对原始问题的回答

错误处理模式​

重试机制​

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=4, max=10))
def call_agent_with_retry(agent, input_text):
    return agent.invoke({"messages": [{"role": "user", "content": input_text}]})

错误处理中间件​

class ErrorHandlingMiddleware(AgentMiddleware):
    def wrap_tool_call(self, request, handler):
        """包装工具调用，处理错误"""
        try:
            result = yield handler(request)
            return result
        except Exception as e:
            tool_name = request.get("name", "unknown")
            return f"工具 {tool_name} 执行失败，已记录日志。"

规划与任务分解​

任务分解示例​

from pydantic import BaseModel, Field
from typing import List

class TaskPlan(BaseModel):
    """任务计划"""
    goal: str = Field(description="原始目标")
    subtasks: List[str] = Field(description="子任务列表")

def decompose_task(goal: str) -> TaskPlan:
    """分解任务"""
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    
    prompt = f"""请将以下目标分解为具体的执行步骤：
    目标：{goal}
    
    返回 JSON 格式的任务列表。"""
    
    chain = llm.with_structured_output(TaskPlan)
    return chain.invoke({"goal": goal})

Plan-and-Execute 模式​

from langgraph.graph import StateGraph, END

# 规划节点
def plan_node(state):
    # 生成执行计划
    return {"plan": [...]}

# 执行节点
def execute_node(state):
    # 执行当前步骤
    return {"current_step": state["current_step"] + 1}

# 构建工作流
workflow = StateGraph(State)
workflow.add_node("plan", plan_node)
workflow.add_node("execute", execute_node)
workflow.set_entry_point("plan")
workflow.add_edge("plan", "execute")
workflow.add_conditional_edges("execute", should_continue, {
    "execute": "execute",
    "end": END
})

思维链推理​

def chain_of_thought(question: str) -> str:
    """使用思维链进行推理"""
    prompt = f"""请一步步思考解决以下问题：
    
    问题：{question}
    
    让我们一步步思考："""
    
    return llm.invoke(prompt).content

性能优化​

并行工具调用​

import asyncio

async def execute_parallel(tool_calls):
    tasks = [execute_tool(call) for call in tool_calls]
    return await asyncio.gather(*tasks)

缓存结果​

from functools import lru_cache

@lru_cache(maxsize=100)
def cached_search(query: str) -> str:
    return search_api(query)

调试技巧​

启用详细日志​

import logging
logging.basicConfig(level=logging.DEBUG)

agent = create_agent(
    model=llm,
    tools=[...],
    verbose=True
)

追踪执行​

# LangSmith 自动追踪
os.environ["LANGSMITH_TRACING"] = "true"

# 查看执行详情
for step in agent.stream({...}):
    print(step)

常见问题解决​

问题解决方案
Agent 陷入循环设置迭代限制，添加终止条件
工具调用参数错误添加参数验证和默认值
上下文过长使用总结或滑动窗口记忆
响应太慢使用流式输出、并行执行
结果不稳定降低 temperature 参数
工具执行失败添加重试机制和错误处理

最佳实践清单​

  • 为每个工具编写清晰的描述
  • 实现健壮的错误处理
  • 添加输入验证
  • 设置合理的迭代限制
  • 使用 LangSmith 追踪执行
  • 对敏感信息进行脱敏
  • 编写单元测试和评估
  • 配置监控和告警

参考链接​