AI Agent 实战练习

什么是 AI Agent？

AI Agent 是能够感知环境、自主决策并执行任务的智能系统。与传统 AI 不同，AI Agent 不是被动响应输入，而是能够主动规划、执行多步骤任务，并调用外部工具。

传统 AI: 输入 → 输出（一次性的）
AI Agent: 输入 → 思考 → 行动 → 观察 → 思考...（循环迭代）

---

核心架构

AI Agent 由四个核心组件构成：

组件	作用	本项目实现
Model（模型）	理解用户输入，生成决策	MiniMax M2
Orchestration（编排）	管理任务流程和执行逻辑	LangChain AgentExecutor
Tools（工具）	扩展 Agent 能力	rag_search, calculator, time_query
Memory（记忆）	存储历史信息	短期记忆 + 长期记忆

---

前置要求

1. 环境准备

# Python 3.11+
python3.11 --version

# 创建虚拟环境（推荐）
python3.11 -m venv venv
source venv/bin/activate  # macOS/Linux

2. 安装依赖

pip install langchain langchain-core langchain-community langchain-openai
pip install llama-index llama-index-llms-openai-like llama-index-embeddings-huggingface
pip install langchain-huggingface  # embedding（如果兼容）
pip install faiss-cpu  # 向量数据库
pip install sentence-transformers  # 嵌入模型
pip install pendulum  # 时间查询（可选）
pip install rich  # 输出美化（可选）

3. API 准备

本项目使用 MiniMax API，你需要：

• API Key
• API Base URL
• 模型名称

---

完整代码

import warnings
warnings.filterwarnings("ignore", category=DeprecationWarning)

import os
os.environ["HF_ENDPOINT"] = "https://hf-mirror.com"

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

from llama_index.core import VectorStoreIndex, Document, Settings
from llama_index.llms.openai_like import OpenAILike
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from langchain.memory import ConversationBufferMemory
from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import HuggingFaceEmbeddings

# ==================== LLM 配置 ====================

API_KEY = "your-api-key"
API_BASE = "https://api.minimax.chat/v1"
MODEL = "MiniMax-M2.7"

llm = ChatOpenAI(
    temperature=0.3,
    model=MODEL,
    openai_api_key=API_KEY,
    openai_api_base=API_BASE
)

# RAG 专用 LLM
rag_llm = OpenAILike(
    model=MODEL,
    api_base=API_BASE,
    api_key=API_KEY,
    is_chat_model=True
)

# ==================== Embedding 配置 ====================

embed_model = HuggingFaceEmbedding(
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)

# LangChain 和 LlamaIndex 需要不同的 embedding 实例
langchain_embed_model = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)

Settings.embed_model = embed_model

# ==================== RAG 文档数据 ====================

documents = [
    Document(text="AI Agent 是能够感知环境、自主决策并执行任务的智能系统。"),
    # ... 更多文档
]

# ==================== 构建索引 ====================

index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine(llm=rag_llm)

# ==================== 定义工具 ====================

@tool
def rag_search(query: str) -> str:
    """查询内部知识库(RAG)"""
    response = query_engine.query(query)
    return str(response)

@tool
def calculator(expression: str) -> str:
    """数学计算器"""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"计算错误: {str(e)}"

@tool
def time_query(city: str) -> str:
    """查询城市时间"""
    from datetime import datetime
    try:
        import pendulum
        dt = pendulum.now(city)
        return dt.to_formatted_date_string() + " " + dt.to_time_string()
    except Exception:
        return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

tools = [rag_search, calculator, time_query]

# ==================== Prompt 模板 ====================

prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个AI Agent助手...

【可用工具】
1. rag_search: 查询知识库
2. calculator: 数学计算
3. time_query: 查询时间

【调用规则】
1. 知识性问题 → rag_search
2. 数学计算 → calculator
3. 时间问题 → time_query

【输出规则】
直接输出最终答案...
"""),
    ("placeholder", "{chat_history}"),
    ("user", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

# ==================== 记忆系统 ====================

# 短期记忆（内存）
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

# 长期记忆（磁盘）
MEMORY_DIR = "./memory_store"
os.makedirs(MEMORY_DIR, exist_ok=True)

def load_vector_memory():
    """加载或创建向量记忆库"""
    index_path = os.path.join(MEMORY_DIR, "index.faiss")
    pkl_path = os.path.join(MEMORY_DIR, "index.pkl")
    if os.path.exists(index_path) and os.path.exists(pkl_path):
        vectorstore = FAISS.load_local(
            MEMORY_DIR,
            langchain_embed_model,
            allow_dangerous_deserialization=True
        )
        print("记忆已从磁盘加载")
    else:
        vectorstore = FAISS.from_texts(
            ["初始化记忆"],
            langchain_embed_model
        )
        print("创建新的记忆库")
    return vectorstore

vectorstore = load_vector_memory()

def save_to_long_memory(user_input, output):
    """保存对话到长期记忆"""
    vectorstore.add_texts([
        f"User: {user_input} | Assistant: {output}"
    ])
    vectorstore.save_local(MEMORY_DIR)

# ==================== 创建 Agent ====================

agent = create_tool_calling_agent(
    llm=llm,
    tools=tools,
    prompt=prompt
)

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

# ==================== 运行 ====================

print("Agent started")

while True:
    user_input = input("You: ")
    if user_input.lower() in ["exit", "quit"]:
        break

    # 1. 检索长期记忆
    relevant_memory = vectorstore.similarity_search(user_input, k=3)
    memory_text = "\n".join([doc.page_content for doc in relevant_memory])

    # 2. 注入 prompt
    full_input = f"历史相关记忆:\n{memory_text}\n当前问题:\n{user_input}"

    # 3. 执行
    result = agent_executor.invoke({"input": full_input})
    output = result["output"]
    print("Agent:", output)

    # 4. 保存长期记忆
    save_to_long_memory(user_input, output)

---

组件详解

1. LLM 配置

llm = ChatOpenAI(
    temperature=0.3,        # 0=确定性，1=创造性
    model=MODEL,
    openai_api_key=API_KEY,
    openai_api_base=API_BASE
)

关键参数：

• temperature: 控制输出随机性。0.3 适合需要准确答案的场景
• openai_api_base: 不同 API 服务商有不同的 base URL

2. 工具定义

使用 @tool 装饰器定义工具：

@tool
def calculator(expression: str) -> str:
    """数学计算器

    Args:
        expression: 数学表达式，如 "2+3*5"

    Returns:
        计算结果字符串
    """
    try:
        return str(eval(expression))
    except Exception as e:
        return f"计算错误: {str(e)}"

工具描述的重要性：

• 描述越详细，LLM 越能正确判断何时调用
• 必须包含：适用场景、不适用场景、参数说明、返回值格式

3. Prompt 模板

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个AI Agent助手..."),
    ("placeholder", "{chat_history}"),  # 短期记忆注入点
    ("user", "{input}"),
    ("placeholder", "{agent_scratchpad}")  # Agent 内部工作区
])

placeholder 类型：

• {input}: 用户输入
• {chat_history}: 短期记忆（ConversationBufferMemory）
• {agent_scratchpad}: Agent 内部推理过程

4. 记忆系统

短期记忆：

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True  # 返回 message 对象列表
)

• 存储当前会话的上下文
• 由 AgentExecutor 自动管理
• 每次对话后自动更新

长期记忆：

vectorstore = FAISS.from_texts(["初始化记忆"], embed_model)
vectorstore.save_local(MEMORY_DIR)  # 持久化
vectorstore = FAISS.load_local(...)  # 加载

• 使用向量数据库存储
• 基于语义相似度检索
• 持久化到磁盘，重启后不丢失

5. Agent 创建

agent = create_tool_calling_agent(
    llm=llm,
    tools=tools,
    prompt=prompt
)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=3,  # 最大迭代次数，防止死循环
    memory=memory
)

关键参数：

• max_iterations: 防止 Agent 在工具调用循环中卡住
• memory: 短期记忆管理器
• verbose: 是否输出调试信息

---

完整项目结构

AI Agent/
├── AI Agent.py          # 主程序
└── memory_store/        # 长期记忆存储目录
    ├── index.faiss      # FAISS 索引文件
    └── index.pkl        # 文档 pickle 文件

---

测试验证

测试 1：基础工具调用

You: AI Agent 有什么应用场景?
Agent: [应调用 rag_search，返回知识库中的答案]

测试 2：数学计算

You: 帮我计算 25*17+99
Agent: [应调用 calculator，输出 524]

测试 3：短期记忆

You: 帮我计算 25*17+99
Agent: 524

You: 把刚才的结果乘以2
Agent: [应记住 524，输出 1048]

测试 4：长期记忆（重启后）

# 重启程序

You: 我们刚才做了什么运算?
Agent: [应检索到之前的 25*17+99=524]

---

常见问题

1. 长期记忆没有持久化

检查：

• save_to_long_memory() 是否在每次对话后调用
• load_vector_memory() 是否在启动时调用
• vectorstore.save_local() 是否执行

2. 工具调用失败

检查：

• 工具描述是否清晰
• 参数格式是否正确
• max_iterations 是否足够

3. 短期记忆不生效

检查：

• prompt 是否包含 {chat_history} placeholder
• return_messages 是否为 True
• AgentExecutor 是否传入了 memory 参数

---

进阶方向

1. Multi-Tools 协作

添加更多工具（搜索、数据库、API 调用），让 Agent 能够组合使用多个工具完成任务。

2. 记忆优化

• 短期记忆：设置 token 限制，避免超出 context window
• 长期记忆：添加时间戳，支持按时间检索

3. Multi-Agent 系统

多个专业 Agent 协同工作，如一个协调者 + 多个执行者。

4. 评估与监控

添加工具调用成功率、任务完成率等指标。

---

关键概念回顾

概念	说明
AI Agent	自主决策、执行任务的智能系统
Tool Calling	LLM 调用外部工具的机制
ReAct	Thought → Action → Observation 循环
短期记忆	ConversationBufferMemory，会话级
长期记忆	FAISS 向量数据库，持久化
RAG	检索增强生成，结合知识库回答问题