跳转到内容

AI Agent 开发入门实战 2026

AI Agent 开发入门实战

🤖 AI Agent 是 AI 应用的下一个范式。 与传统的"提问-回答"模式不同,Agent 能够自主规划任务、调用外部工具、根据反馈调整策略,完成复杂的多步骤工作。2026 年,Agent 框架已经成熟,本文带你从零构建自己的 AI Agent。

本文将带你系统掌握:

  • ✅ AI Agent 核心概念与架构
  • ✅ Agent 与普通 LLM 调用的区别
  • ✅ 工具调用(Function Calling)原理与实战
  • ✅ LangChain Agent 开发
  • ✅ 记忆系统与上下文管理
  • ✅ RAG 检索增强生成
  • ✅ 多 Agent 协作
  • ✅ 生产部署与最佳实践

一、AI Agent 核心概念

1.1 什么是 AI Agent

AI Agent 是一个能够自主感知、决策和执行的智能体:

传统 LLM:用户提问 → LLM 回答
AI Agent:用户目标 → 规划任务 → 调用工具 → 观察结果 → 调整策略 → 完成目标

Agent 的核心能力:

  • 感知(Perception):理解用户意图和环境信息
  • 规划(Planning):将目标分解为可执行的步骤
  • 执行(Action):调用工具/API 完成具体任务
  • 反思(Reflection):根据执行结果调整策略

1.2 Agent vs 普通 LLM

特性普通 LLMAI Agent
交互模式单轮问答多轮自主执行
工具使用可调用外部 API
记忆仅当前上下文短期+长期记忆
规划能力自动分解任务
错误处理无法重试自动调整重试
适用场景简单问答复杂多步任务

1.3 Agent 架构

┌─────────────────────────────────────┐
│            AI Agent                 │
│  ┌───────────┐  ┌───────────────┐  │
│  │  LLM 大脑 │  │   记忆系统    │  │
│  └─────┬─────┘  └───────┬───────┘  │
│        │                │          │
│  ┌─────▼─────────────────▼───────┐  │
│  │        规划与决策引擎        │  │
│  └─────┬─────────────────┬───────┘  │
│        │                │          │
│  ┌─────▼─────┐  ┌───────▼───────┐  │
│  │  工具调用  │  │  观察与反思  │  │
│  └───────────┘  └───────────────┘  │
└─────────────────────────────────────┘

┌────────▼────────────────────────────┐
│          外部工具 / API             │
│  搜索 │ 数据库 │ 文件 │ 代码执行   │
└─────────────────────────────────────┘

二、环境准备

2.1 安装依赖

bash
# 创建项目
mkdir ai-agent-demo && cd ai-agent-demo
python -m venv venv
source venv/bin/activate

# 安装 LangChain
pip install langchain langchain-openai langchain-community

# 安装 LlamaIndex
pip install llama-index llama-index-llms-openai

# 安装工具依赖
pip install wikipedia duckduckgo-search

2.2 配置 API Key

bash
# 设置环境变量
export OPENAI_API_KEY="sk-xxx"

# 或使用 .env 文件
echo 'OPENAI_API_KEY=sk-xxx' > .env

💡 也可以使用其他模型提供商:Anthropic、DeepSeek、本地 Ollama 等。


三、工具调用(Function Calling)

3.1 什么是工具调用

工具调用让 LLM 能够"使用"外部工具:

  1. LLM 根据用户请求决定需要调用哪个工具
  2. 返回工具名称和参数
  3. 执行工具获得结果
  4. 将结果反馈给 LLM 继续推理

3.2 OpenAI Function Calling

python
from openai import OpenAI

client = OpenAI()

# 定义工具
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "执行数学计算",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式"
                    }
                },
                "required": ["expression"]
            }
        }
    }
]

# 工具实现
def get_weather(city: str) -> str:
    # 实际应用中调用天气 API
    weather_data = {
        "北京": "晴天,25°C,湿度45%",
        "上海": "多云,28°C,湿度65%",
        "深圳": "小雨,30°C,湿度80%"
    }
    return weather_data.get(city, f"{city}: 暂无数据")

def calculate(expression: str) -> str:
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"计算错误: {e}"

# 调用 LLM
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "北京今天天气怎么样?顺便算一下 25 * 4 + 10"}
    ],
    tools=tools,
    tool_choice="auto"
)

# 处理工具调用
message = response.choices[0].message

if message.tool_calls:
    for tool_call in message.tool_calls:
        func_name = tool_call.function.name
        func_args = json.loads(tool_call.function.arguments)
        
        if func_name == "get_weather":
            result = get_weather(**func_args)
        elif func_name == "calculate":
            result = calculate(**func_args)
        
        print(f"调用 {func_name}({func_args}) = {result}")

四、LangChain Agent 开发

4.1 使用 LangChain 工具

python
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.tools import DuckDuckGoSearchRun, WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper

# 初始化 LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 定义工具
search = DuckDuckGoSearchRun()
wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())

# 自定义工具
from langchain_core.tools import tool

@tool
def get_current_time() -> str:
    """获取当前时间"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

@tool
def calculate(expression: str) -> str:
    """执行数学计算,expression 为数学表达式"""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"计算错误: {e}"

tools = [search, wikipedia, get_current_time, calculate]

4.2 创建 Agent

python
# 创建提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个智能助手,可以使用工具来帮助用户完成任务。"),
    ("user", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

# 创建 Agent
agent = create_tool_calling_agent(llm, tools, prompt)

# 创建 Agent 执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=5
)

# 运行 Agent
result = agent_executor.invoke({
    "input": "搜索 LangChain 的最新版本信息,并告诉我当前时间"
})

print(result["output"])

4.3 ReAct Agent

ReAct(Reasoning + Acting)是最经典的 Agent 模式:

python
from langchain.agents import create_react_agent

prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个智能助手,按照以下格式思考和行动:

Question: 用户的问题
Thought: 你应该怎么思考
Action: 要使用的工具名称
Action Input: 工具的输入参数
Observation: 工具的返回结果
... (Thought/Action/Action Input/Observation 可以重复多次)
Thought: 我现在知道最终答案了
Final Answer: 最终答案"""),
    ("user", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

五、记忆系统

5.1 对话记忆

python
from langchain.memory import ConversationBufferMemory

# 基础记忆(保存完整对话)
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

# 窗口记忆(只保留最近 K 轮)
from langchain.memory import ConversationBufferWindowMemory
memory = ConversationBufferWindowMemory(
    k=5,  # 保留最近 5 轮
    memory_key="chat_history",
    return_messages=True
)

# 摘要记忆(对历史对话做摘要)
from langchain.memory import ConversationSummaryMemory
memory = ConversationSummaryMemory(
    llm=llm,
    memory_key="chat_history"
)

5.2 长期记忆

python
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# 向量存储作为长期记忆
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(
    embedding_function=embeddings,
    persist_directory="./memory_db"
)

# 存储重要信息
vectorstore.add_texts(["用户喜欢 Python", "用户是后端开发者"])

# 检索相关记忆
relevant_memory = vectorstore.similarity_search("编程语言偏好", k=3)

六、RAG 检索增强生成

6.1 RAG 原理

用户提问 → 检索相关文档 → 将文档注入 Prompt → LLM 生成回答

6.2 使用 LlamaIndex 实现 RAG

python
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

# 配置模型
Settings.llm = OpenAI(model="gpt-4o")
Settings.embed_model = OpenAIEmbedding()

# 加载文档
documents = SimpleDirectoryReader("./data").load_data()

# 构建索引
index = VectorStoreIndex.from_documents(documents)

# 创建查询引擎
query_engine = index.as_query_engine()

# 查询
response = query_engine.query("LangChain 的主要组件有哪些?")
print(response)

6.3 使用 LangChain 实现 RAG

python
from langchain_community.document_loaders import DirectoryLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.chains import RetrievalQA

# 加载文档
loader = DirectoryLoader("./data", glob="**/*.md")
docs = loader.load()

# 分割文档
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
chunks = splitter.split_documents(docs)

# 构建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 创建 RAG 链
llm = ChatOpenAI(model="gpt-4o")
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3})
)

# 查询
result = qa_chain.invoke("什么是 AI Agent?")
print(result["result"])

七、多 Agent 协作

7.1 Agent 角色分工

python
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate

# 研究员 Agent
researcher_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个研究员,负责搜索和收集信息。"),
    ("user", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])
researcher = create_tool_calling_agent(llm, [search, wikipedia], researcher_prompt)
researcher_executor = AgentExecutor(agent=researcher, tools=[search, wikipedia])

# 写作者 Agent
writer_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个技术写作者,根据研究结果撰写文章。"),
    ("user", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])
writer = create_tool_calling_agent(llm, [], writer_prompt)
writer_executor = AgentExecutor(agent=writer, tools=[])

# 协作流程
research_result = researcher_executor.invoke({"input": "研究 AI Agent 最新进展"})
article = writer_executor.invoke({"input": f"根据以下研究写一篇文章:\n{research_result['output']}"})
print(article["output"])

7.2 使用 LangGraph 构建复杂工作流

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

# 定义状态
class AgentState(TypedDict):
    query: str
    research_result: str
    draft: str
    review_feedback: str
    final_output: str

# 定义节点
def research_node(state: AgentState) -> AgentState:
    result = researcher_executor.invoke({"input": state["query"]})
    state["research_result"] = result["output"]
    return state

def write_node(state: AgentState) -> AgentState:
    prompt = f"根据研究写文章:{state['research_result']}"
    result = writer_executor.invoke({"input": prompt})
    state["draft"] = result["output"]
    return state

def review_node(state: AgentState) -> AgentState:
    # 审核逻辑
    state["review_feedback"] = "文章质量良好"
    return state

# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("research", research_node)
workflow.add_node("write", write_node)
workflow.add_node("review", review_node)

workflow.set_entry_point("research")
workflow.add_edge("research", "write")
workflow.add_edge("write", "review")
workflow.add_edge("review", END)

app = workflow.compile()
result = app.invoke({"query": "写一篇关于 AI Agent 的技术文章"})

八、实战:构建个人助手 Agent

8.1 完整代码

python
#!/usr/bin/env python3
"""个人助手 Agent - 支持搜索、计算、文件操作"""

import os
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from langchain.memory import ConversationBufferWindowMemory
from langchain_community.tools import DuckDuckGoSearchRun

# 初始化
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 工具定义
search = DuckDuckGoSearchRun()

@tool
def calculate(expression: str) -> str:
    """执行数学计算"""
    try:
        allowed = set("0123456789+-*/.() ")
        if not all(c in allowed for c in expression):
            return "不安全的表达式"
        return str(eval(expression))
    except Exception as e:
        return f"错误: {e}"

@tool
def read_file(file_path: str) -> str:
    """读取文件内容"""
    try:
        with open(file_path, "r") as f:
            return f.read()[:5000]  # 限制长度
    except Exception as e:
        return f"读取失败: {e}"

@tool
def list_directory(dir_path: str = ".") -> str:
    """列出目录内容"""
    try:
        items = os.listdir(dir_path)
        return "\n".join(items[:50])
    except Exception as e:
        return f"列出失败: {e}"

tools = [search, calculate, read_file, list_directory]

# 创建 Agent
prompt = ChatPromptTemplate.from_messages([
    ("system", """你是个人智能助手,可以帮助用户:
    1. 搜索信息
    2. 数学计算
    3. 读取文件
    4. 管理目录
    请根据用户需求选择合适的工具。"""),
    ("placeholder", "{chat_history}"),
    ("user", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

memory = ConversationBufferWindowMemory(
    k=5,
    memory_key="chat_history",
    return_messages=True
)

agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True
)

# 交互循环
while True:
    user_input = input("\n你: ")
    if user_input.lower() in ["exit", "quit"]:
        break
    result = agent_executor.invoke({"input": user_input})
    print(f"\n助手: {result['output']}")

九、生产部署最佳实践

9.1 成本控制

  • 缓存结果:相同请求返回缓存
  • 限制 Token:设置 max_tokens 防止超长输出
  • 选择合适的模型:简单任务用小模型,复杂任务用大模型
  • 批量处理:合并多个请求

9.2 可靠性

  • 超时设置:防止 LLM 调用无限等待
  • 重试机制:网络错误自动重试
  • 降级策略:主模型不可用时切换备用模型
  • 输入验证:清洗用户输入,防止注入攻击

9.3 可观测性

python
from langchain.callbacks import StdOutCallbackHandler

# 使用回调追踪
handler = StdOutCallbackHandler()
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callbacks=[handler],
    verbose=True
)

9.4 安全建议

建议说明
限制工具权限只给 Agent 必要的最小权限
审计日志记录所有工具调用
人工审批危险操作需人工确认
输入过滤防止 Prompt 注入
输出审查检查 LLM 输出的安全性

十、常用框架对比

框架特点适用场景
LangChain生态丰富,组件全面通用 Agent 开发
LlamaIndexRAG 能力强,数据索引知识库问答
CrewAI多 Agent 协作团队协作场景
AutoGen微软出品,对话式多轮对话
Semantic Kernel微软企业级.NET/企业应用
Dify低代码,可视化快速原型

十一、总结

AI Agent 开发是 2026 年最热门的方向之一:

  1. 从单次问答到自主执行:Agent 能完成复杂的多步任务
  2. 工具调用是核心:让 LLM 与外部世界连接
  3. 记忆系统是关键:短期记忆+长期记忆实现持续对话
  4. RAG 解决知识问题:让 Agent 获取最新和私有知识
  5. 多 Agent 协作是趋势:专业分工,高效协作

🤖 掌握 AI Agent 开发,你就掌握了 AI 应用的未来。


相关文章推荐: