Tutorial 8: Agents 与 Tools
Agent 概述
Agent 是能够自主决策和使用工具的智能系统。 LlamaIndex 提供了构建 Agent 的完整框架。
Agent 工作流程：
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ 用户输入 │───►│ Agent │───►│ 选择工具 │
└─────────────┘ │ (推理) │ └─────────────┘
└─────────────┘ │
▲ ▼
│ ┌─────────────┐
│ │ 执行工具 │
│ └─────────────┘
│ │
┌─────────────┐ │
│ 观察结果 │◄──────────┘
└─────────────┘
│
▼
┌─────────────┐
│ 生成回答 │
└─────────────┘
ReAct Agent
基于 ReAct（Reasoning + Acting）范式的 Agent。
基本用法
from llama_index.core.agent import ReActAgent
from llama_index.core.tools import FunctionTool
from llama_index.llms.openai import OpenAI
# 定义工具函数
def multiply(a: int, b: int) -> int:
"""将两个整数相乘"""
return a * b
def add(a: int, b: int) -> int:
"""将两个整数相加"""
return a + b
# 创建工具
multiply_tool = FunctionTool.from_defaults(fn=multiply)
add_tool = FunctionTool.from_defaults(fn=add)
# 创建 Agent
llm = OpenAI(model="gpt-4o-mini")
agent = ReActAgent.from_tools(
[multiply_tool, add_tool],
llm=llm,
verbose=True
)
# 使用 Agent
response = agent.chat("计算 (3 + 5) * 2 的结果")
print(response)
查看推理过程
# Agent 会显示其推理过程（当 verbose=True）
# Thought: 我需要先计算 3 + 5
# Action: add
# Action Input: {"a": 3, "b": 5}
# Observation: 8
# Thought: 现在我需要将结果乘以 2
# Action: multiply
# Action Input: {"a": 8, "b": 2}
# Observation: 16
# Thought: 我现在知道最终答案了
# Answer: (3 + 5) * 2 = 16
FunctionTool
将 Python 函数转换为 Agent 可用的工具。
基本用法
from llama_index.core.tools import FunctionTool
def get_weather(city: str) -> str:
"""获取指定城市的天气信息
Args:
city: 城市名称
"""
# 模拟天气API
weather_data = {
"北京": "晴天，25°C",
"上海": "多云，28°C",
"深圳": "小雨，30°C"
}
return weather_data.get(city, f"未找到{city}的天气信息")
# 创建工具
weather_tool = FunctionTool.from_defaults(
fn=get_weather,
name="get_weather",
description="获取指定城市的当前天气"
)
# 使用
agent = ReActAgent.from_tools([weather_tool], llm=llm, verbose=True)
response = agent.chat("北京今天天气怎么样？")
异步工具
import aiohttp
from llama_index.core.tools import FunctionTool
async def async_fetch_data(url: str) -> str:
"""异步获取URL内容"""
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()
async_tool = FunctionTool.from_defaults(
async_fn=async_fetch_data,
name="fetch_url",
description="获取URL的内容"
)
QueryEngineTool
将查询引擎包装为工具。
from llama_index.core import VectorStoreIndex, Document
from llama_index.core.tools import QueryEngineTool, ToolMetadata
# 创建知识库索引
documents = [
Document(text="公司成立于2020年，总部位于北京。"),
Document(text="公司主要产品是AI解决方案，服务于企业客户。"),
Document(text="公司2023年营收10亿元，同比增长50%。"),
]
index = VectorStoreIndex.from_documents(documents)
# 创建查询引擎工具
query_engine = index.as_query_engine()
query_tool = QueryEngineTool(
query_engine=query_engine,
metadata=ToolMetadata(
name="company_knowledge",
description="查询公司相关信息，包括历史、产品、财务等"
)
)
# 创建 Agent
agent = ReActAgent.from_tools([query_tool], llm=llm, verbose=True)
response = agent.chat("公司的营收情况如何？")
print(response)
多知识库 Agent
# 创建多个知识库
product_docs = [Document(text="产品A是我们的旗舰产品...")]
finance_docs = [Document(text="2023年财务报告显示...")]
hr_docs = [Document(text="公司目前有500名员工...")]
product_index = VectorStoreIndex.from_documents(product_docs)
finance_index = VectorStoreIndex.from_documents(finance_docs)
hr_index = VectorStoreIndex.from_documents(hr_docs)
# 创建多个查询工具
tools = [
QueryEngineTool(
query_engine=product_index.as_query_engine(),
metadata=ToolMetadata(
name="product_info",
description="查询产品相关信息"
)
),
QueryEngineTool(
query_engine=finance_index.as_query_engine(),
metadata=ToolMetadata(
name="finance_info",
description="查询财务相关信息"
)
),
QueryEngineTool(
query_engine=hr_index.as_query_engine(),
metadata=ToolMetadata(
name="hr_info",
description="查询人力资源相关信息"
)
),
]
# 创建多知识库 Agent
agent = ReActAgent.from_tools(tools, llm=llm, verbose=True)
# Agent 会自动选择合适的知识库
response = agent.chat("公司的员工人数是多少？")
自定义工具
创建更复杂的自定义工具。
from llama_index.core.tools import BaseTool, ToolMetadata, ToolOutput
from typing import Any
class DatabaseTool(BaseTool):
"""数据库查询工具"""
def __init__(self, connection_string: str):
self._connection_string = connection_string
@property
def metadata(self) -> ToolMetadata:
return ToolMetadata(
name="database_query",
description="执行SQL查询获取数据库信息"
)
def call(self, query: str) -> ToolOutput:
"""执行查询"""
# 这里是模拟，实际应该连接数据库
result = f"执行查询: {query}"
return ToolOutput(
content=result,
tool_name=self.metadata.name,
raw_input={"query": query},
raw_output=result
)
# 使用自定义工具
db_tool = DatabaseTool("postgresql://localhost/mydb")
agent = ReActAgent.from_tools([db_tool], llm=llm)
工具规范
from llama_index.core.tools import ToolSpec
class SearchToolSpec(ToolSpec):
"""搜索工具规范"""
spec_functions = ["web_search", "image_search"]
def web_search(self, query: str) -> str:
"""搜索网页
Args:
query: 搜索关键词
"""
return f"网页搜索结果: {query}"
def image_search(self, query: str) -> str:
"""搜索图片
Args:
query: 搜索关键词
"""
return f"图片搜索结果: {query}"
# 从规范创建工具
search_spec = SearchToolSpec()
tools = search_spec.to_tool_list()
agent = ReActAgent.from_tools(tools, llm=llm)
内置工具
LlamaIndex 提供了一些内置工具。
代码解释器
# pip install llama-index-tools-code-interpreter
from llama_index.tools.code_interpreter import CodeInterpreterToolSpec
code_spec = CodeInterpreterToolSpec()
tools = code_spec.to_tool_list()
agent = ReActAgent.from_tools(tools, llm=llm, verbose=True)
response = agent.chat("计算1到100的素数之和")
维基百科工具
# pip install llama-index-tools-wikipedia
from llama_index.tools.wikipedia import WikipediaToolSpec
wiki_spec = WikipediaToolSpec()
tools = wiki_spec.to_tool_list()
agent = ReActAgent.from_tools(tools, llm=llm, verbose=True)
response = agent.chat("介绍一下阿尔伯特·爱因斯坦")
会话历史
Agent 支持多轮对话。
from llama_index.core.agent import ReActAgent
from llama_index.core.memory import ChatMemoryBuffer
# 创建带记忆的 Agent
memory = ChatMemoryBuffer.from_defaults(token_limit=3000)
agent = ReActAgent.from_tools(
tools,
llm=llm,
memory=memory,
verbose=True
)
# 多轮对话
response1 = agent.chat("公司的主要产品是什么？")
print(f"回答1: {response1}")
response2 = agent.chat("它的价格是多少？") # 会记住上下文
print(f"回答2: {response2}")
# 查看对话历史
print("\n对话历史:")
for msg in agent.chat_history:
print(f"{msg.role}: {msg.content[:50]}...")
# 重置对话
agent.reset()
OpenAI Agent
使用 OpenAI 的 Function Calling 能力。
from llama_index.agent.openai import OpenAIAgent
from llama_index.llms.openai import OpenAI
# 创建 OpenAI Agent
llm = OpenAI(model="gpt-4o-mini")
agent = OpenAIAgent.from_tools(
tools,
llm=llm,
verbose=True
)
response = agent.chat("查询公司的财务信息")
print(response)
流式响应
# 流式对话
streaming_response = agent.stream_chat("详细介绍公司的产品线")
for token in streaming_response.response_gen:
print(token, end="", flush=True)
并行工具调用
同时调用多个工具。
from llama_index.agent.openai import OpenAIAgent
# OpenAI Agent 支持并行函数调用
agent = OpenAIAgent.from_tools(
tools,
llm=OpenAI(model="gpt-4o"),
verbose=True
)
# 当问题涉及多个领域时，Agent 可能并行调用多个工具
response = agent.chat("告诉我公司的产品和财务状况")
Agent 执行器
自定义 Agent 的执行逻辑。
from llama_index.core.agent import AgentRunner
from llama_index.core.agent.react.step import ReActAgentWorker
# 创建 Agent Worker
agent_worker = ReActAgentWorker.from_tools(
tools,
llm=llm,
verbose=True
)
# 创建 Agent Runner
agent = AgentRunner(agent_worker)
# 创建任务
task = agent.create_task("分析公司的财务状况")
# 逐步执行
step_output = agent.run_step(task.task_id)
print(f"步骤完成: {step_output.is_last}")
# 继续执行直到完成
while not step_output.is_last:
step_output = agent.run_step(task.task_id)
print(f"步骤输出: {step_output.output}")
# 获取最终响应
response = agent.finalize_response(task.task_id)
print(f"最终答案: {response}")
实战示例
构建一个智能助手 Agent。
from llama_index.core.agent import ReActAgent
from llama_index.core.tools import FunctionTool, QueryEngineTool, ToolMetadata
from llama_index.core import VectorStoreIndex, Document
from llama_index.llms.openai import OpenAI
from datetime import datetime
import json
class SmartAssistant:
"""智能助手"""
def __init__(self):
self.llm = OpenAI(model="gpt-4o-mini")
self.tools = []
self._setup_tools()
self.agent = None
self._build_agent()
def _setup_tools(self):
"""设置工具"""
# 1. 时间工具
def get_current_time() -> str:
"""获取当前时间"""
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.tools.append(FunctionTool.from_defaults(
fn=get_current_time,
name="get_time",
description="获取当前日期和时间"
))
# 2. 计算工具
def calculate(expression: str) -> str:
"""计算数学表达式
Args:
expression: 数学表达式，如 "2 + 3 * 4"
"""
try:
result = eval(expression)
return str(result)
except Exception as e:
return f"计算错误: {str(e)}"
self.tools.append(FunctionTool.from_defaults(
fn=calculate,
name="calculator",
description="计算数学表达式"
))
# 3. 笔记工具
self.notes = []
def add_note(content: str) -> str:
"""添加笔记
Args:
content: 笔记内容
"""
self.notes.append({
"content": content,
"time": datetime.now().isoformat()
})
return f"笔记已添加: {content}"
def get_notes() -> str:
"""获取所有笔记"""
if not self.notes:
return "暂无笔记"
return json.dumps(self.notes, ensure_ascii=False, indent=2)
self.tools.append(FunctionTool.from_defaults(
fn=add_note,
name="add_note",
description="添加一条笔记"
))
self.tools.append(FunctionTool.from_defaults(
fn=get_notes,
name="get_notes",
description="获取所有笔记"
))
# 4. 知识库工具
knowledge_docs = [
Document(text="Python是一种解释型、面向对象的高级程序设计语言。"),
Document(text="机器学习是人工智能的一个分支，通过算法从数据中学习。"),
Document(text="深度学习使用多层神经网络，特别适合处理图像和语言。"),
]
knowledge_index = VectorStoreIndex.from_documents(knowledge_docs)
self.tools.append(QueryEngineTool(
query_engine=knowledge_index.as_query_engine(),
metadata=ToolMetadata(
name="knowledge_base",
description="查询技术知识库，包含编程、AI等内容"
)
))
def _build_agent(self):
"""构建 Agent"""
self.agent = ReActAgent.from_tools(
self.tools,
llm=self.llm,
verbose=True,
system_prompt="""你是一个智能助手，可以：
1. 查询当前时间
2. 进行数学计算
3. 管理笔记
4. 查询技术知识
请根据用户的问题选择合适的工具来帮助回答。
"""
)
def chat(self, message: str) -> str:
"""与助手对话"""
response = self.agent.chat(message)
return str(response)
def reset(self):
"""重置对话历史"""
self.agent.reset()
# 使用示例
assistant = SmartAssistant()
# 时间查询
print("=" * 50)
print(assistant.chat("现在几点了？"))
# 计算
print("=" * 50)
print(assistant.chat("帮我计算 (15 * 8 + 32) / 4"))
# 笔记
print("=" * 50)
print(assistant.chat("帮我记一下：明天下午3点开会"))
print(assistant.chat("我有哪些笔记？"))
# 知识查询
print("=" * 50)
print(assistant.chat("什么是深度学习？"))
# 复合任务
print("=" * 50)
print(assistant.chat("现在几点了？另外帮我查一下Python是什么"))
小结
本教程介绍了：
Agent 的基本概念和工作流程
ReAct Agent 的使用
各种工具的创建：FunctionTool、QueryEngineTool
自定义工具和工具规范
内置工具的使用
会话历史和记忆管理
OpenAI Agent 和流式响应
完整的智能助手实现
下一步
在下一个教程中，我们将学习如何构建一个完整的知识库系统， 将前面学到的所有技术整合在一起。
练习
创建一个能够搜索网页的 Agent
实现一个多轮对话的客服 Agent
构建一个能够执行代码的编程助手
创建一个多知识库的企业问答 Agent