AI Agent 知识速查表

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

核心概念

概念	说明
Agent	能够感知环境、做出决策并采取行动的智能系统
LLM	大语言模型，Agent 的"大脑"，负责推理和决策
Tool	工具，Agent 与外部世界交互的桥梁
Memory	记忆系统，存储对话历史和知识信息
Planning	规划模块，将复杂任务分解为子任务
ReAct	推理与行动交替进行的执行模式
Function Calling	LLM 调用外部函数的机制

Agent 公式

AI Agent = LLM + Planning + Memory + Tools

LangChain 常用 API

定义工具

from langchain_core.tools import tool

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

创建 Agent

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

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

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个助手。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    handle_parsing_errors=True,
    max_iterations=10
)

result = agent_executor.invoke({"input": "用户输入"})

添加记忆

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory
)

记忆类型

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

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)
    result = execute_function(function_name, 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}
)

工具设计模式

单一职责

@tool
def search(query: str) -> str:
    """搜索信息"""
    pass

@tool
def save(data: str) -> str:
    """保存数据"""
    pass

输入验证

from pydantic import BaseModel, Field, validator

class ToolInput(BaseModel):
    email: str = Field(description="邮箱地址")
    
    @validator('email')
    def validate_email(cls, v):
        if '@' not in v:
            raise ValueError('无效的邮箱')
        return v

错误处理

@tool
def safe_tool(param: str) -> str:
    """安全的工具"""
    try:
        result = do_something(param)
        return f"成功：{result}"
    except ValueError as e:
        return f"参数错误：{e}"
    except Exception as e:
        return f"执行失败：{e}"

多 Agent 协作模式

顺序协作

workflow.add_node("agent_a", agent_a)
workflow.add_node("agent_b", agent_b)
workflow.add_node("agent_c", agent_c)

workflow.set_entry_point("agent_a")
workflow.add_edge("agent_a", "agent_b")
workflow.add_edge("agent_b", "agent_c")
workflow.add_edge("agent_c", END)

层级协作

workflow.add_node("supervisor", supervisor)
workflow.add_node("worker_1", worker_1)
workflow.add_node("worker_2", worker_2)

workflow.set_entry_point("supervisor")
workflow.add_conditional_edges("supervisor", route)
workflow.add_edge("worker_1", "supervisor")
workflow.add_edge("worker_2", "supervisor")

常用提示词模板

ReAct 提示词

你是一个智能助手，可以使用工具完成任务。

可用工具：
{tool_names}

使用以下格式：

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

开始！

问题：{input}
思考：{agent_scratchpad}

系统提示词

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

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

性能优化技巧

并行工具调用

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)

流式输出

async def stream_response(agent, input_text):
    async for chunk in agent.astream(input_text):
        yield chunk

调试技巧

启用详细日志

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True  # 打印执行过程
)

记录中间状态

def debug_node(state):
    print(f"当前状态：{state}")
    result = process(state)
    print(f"处理结果：{result}")
    return result

错误追踪

import traceback

@tool
def debug_tool(param: str) -> str:
    try:
        return do_something(param)
    except Exception as e:
        return f"错误：{e}\n堆栈：{traceback.format_exc()}"

常见问题解决

问题	解决方案
Agent 陷入循环	设置 max_iterations 限制
工具调用参数错误	添加参数验证和默认值
上下文过长	使用 ConversationSummaryMemory
响应太慢	使用流式输出、并行执行
结果不稳定	降低 temperature 参数

最佳实践清单

• 为每个工具编写清晰的描述
• 实现健壮的错误处理
• 添加输入验证
• 设置合理的迭代限制
• 使用记忆系统保持上下文
• 对敏感信息进行脱敏
• 记录日志便于调试
• 编写单元测试

参考链接

• LangChain 文档
• LangGraph 文档
• OpenAI API 文档
• ReAct 论文