这是“从头构建AI智能体”系列文章的第一篇。在本系列中,我们将不使用任何大型语言模型(LLM,即 Large Language Model)编排框架,逐步构建 AI 智能体。接下来,让我们看看本篇文章将要介绍的内容:
- 什么是 AI 智能体?工具使用能力的工作原理是什么?如何构建一个装饰器包装器,从 Python 函数中提取关键信息,并通过系统提示传递给 LLM?如何设计有效的系统提示,用于构建智能体?如何实现一个智能体类,能够利用提供的工具进行规划和执行操作?
“AI 的未来是智能体。”
“2025 年将是智能体之年。”
如今,这样的说法不绝于耳,而且不无道理。为了从大型语言模型(LLM)中挖掘最大的商业价值,我们正转向复杂的智能体流程。
什么是 AI 智能体?
从最简单的角度定义,AI 智能体是一个以 LLM 为核心推理引擎的应用,用于决定实现用户意图所需的步骤。通常,AI 智能体被描述为由多个模块组成的应用。让我们通过一个图示来更好地理解 AI 智能体的结构:
AI 智能体
Planning(规划) :智能体能够制定一系列操作步骤,以实现用户提供的意图。
Memory(记忆) :包括短期和长期记忆,用于存储智能体推理所需的信息。这些信息通常通过系统提示传递给 LLM 作为核心的一部分。
Tools (工具) :智能体可以通过调用各种功能来增强其推理能力。让我们进一步了解工具的多样性:
- 代码中定义的简单函数;包含上下文的向量数据库(VectorDB,即 Vector Database)或其他数据存储;常规机器学习模型 API;甚至是其他智能体!
本篇文章将重点介绍工具使用能力的实现。 如果你正在使用某些智能体编排框架,可能对工具使用的底层细节了解不多。本文将帮助你理解智能体如何提供和使用工具的真正含义。理解应用的基础构建模块非常重要,原因如下:
- 框架通常会隐藏系统提示的实现细节,而不同用例可能需要不同的方法。可能需要调整底层细节,以优化智能体的性能。深入理解系统的工作原理,有助于培养系统思维,从而更高效地构建高级应用。
工具使用能力的高层概述
在构建智能体应用时,首先要明白的是,LLM 本身并不运行代码,它仅通过提示生成意图。为什么 ChatGPT 可以浏览互联网并返回更准确、更新的结果?因为 ChatGPT 本身就是一个智能体,其背后隐藏了许多非 LLM 模块,我们通过 API 无法直接看到。
在构建智能体应用时,提示工程(Prompt Engineering)至关重要,尤其是系统提示的设计。简化的提示结构如下图所示:
只有当你能高效地为系统提示提供可用的工具定义和预期输出(以规划操作或直接回答的形式)时,智能体才能表现出色。想象一下,智能体就像一个厨师,工具则是厨房里的各种器具。厨师根据食谱(系统提示)决定使用哪些器具(工具)来完成菜肴(用户意图)。
实现智能体
在本部分,我们将创建一个 AI 智能体。这个智能体能够在线查询货币汇率,并在需要时执行货币转换以回答用户的问题。首先,我们需要准备好代码和相关资源。
准备 Python 函数作为工具
为智能体提供工具的最简单和最便捷的方式是通过函数,在本项目中我们将使用 Python。我们不需要将函数代码本身提供给系统提示,但需要提取函数的相关信息,以便 LLM 决定是否以及如何调用该函数。
我们定义一个数据类(dataclass),包含所需信息以及可运行的函数:
@dataclassclass Tool: name: str description: str func: Callable[..., str] parameters: Dict[str, Dict[str, str]] def __call__(self, *args, **kwargs) -> str: return self.func(*args, **kwargs)
提取的信息包括:
- 函数名称;函数描述(从文档字符串中提取);函数的可调用对象,以便智能体调用;函数参数信息,以便 LLM 决定如何调用函数。
接下来,我们需要从定义的函数中提取上述信息。我们对函数有一个要求:必须有格式规范的文档字符串(docstring),格式如下:
"""工具功能的描述。参数: - param1:第一个参数的描述 - param2:第二个参数的描述"""
以下函数用于提取参数信息——参数名称和描述:
def parse_docstring_params(docstring: str) -> Dict[str, str]: """从文档字符串中提取参数描述。""" if not docstring: return {} params = {} lines = docstring.split('\n') in_params = False current_param = None for line in lines: line = line.strip() if line.startswith('Parameters:'): in_params = True elif in_params: if line.startswith('-') or line.startswith('*'): current_param = line.lstrip('- *').split(':')[0].strip() params[current_param] = line.lstrip('- *').split(':')[1].strip() elif current_param and line: params[current_param] += ' ' + line.strip() elif not line: in_params = False return params
我们还将从函数定义中的类型提示(type hints)中提取参数类型。以下函数帮助格式化这些类型:
def get_type_description(type_hint: Any) -> str: """获取类型提示的可读描述。""" if isinstance(type_hint, _GenericAlias): if type_hint._name == 'Literal': return f"one of {type_hint.__args__}" return type_hint.__name__
将函数转化为工具的一个便捷方式是使用装饰器。以下代码定义了一个工具装饰器,用于包装函数,可以使用函数名作为工具名,或通过装饰器提供自定义名称:
def tool(name: str = None): def decorator(func: Callable[..., str]) -> Tool: tool_name = name or func.__name__ description = inspect.getdoc(func) or "No description available" type_hints = get_type_hints(func) param_docs = parse_docstring_params(description) sig = inspect.signature(func) params = {} for param_name, param in sig.parameters.items(): params[param_name] = { "type": get_type_description(type_hints.get(param_name, Any)), "description": param_docs.get(param_name, "No description available") } return Tool( name=tool_name, description=description.split('\n\n')[0], func=func, parameters=params ) return decorator
货币转换工具
以下代码通过一个函数创建工具,该函数接收待转换的货币金额、源货币代码和目标货币代码,查询最新的汇率并计算转换后的金额。代码使用 API https://open.er-api.com/v6/latest/{from_currency.upper()}
获取最新汇率:
@tool()def convert_currency(amount: float, from_currency: str, to_currency: str) -> str: """使用最新汇率转换货币。 参数: - amount:待转换的金额 - from_currency:源货币代码(如 USD) - to_currency:目标货币代码(如 EUR) """ try: url = f"https://open.er-api.com/v6/latest/{from_currency.upper()}" with urllib.request.urlopen(url) as response: data = json.loads(response.read()) if "rates" not in data: return "错误:无法获取汇率数据" rate = data["rates"].get(to_currency.upper()) if not rate: return f"错误:未找到 {to_currency} 的汇率" converted = amount * rate return f"{amount} {from_currency.upper()} = {converted:.2f} {to_currency.upper()}" except Exception as e: return f"货币转换错误:{str(e)}"
让我们运行一下:
convert_currency
输出应类似于:
Tool(name='convert_currency', description='使用最新汇率转换货币。', func=<function convert_currency at 0x106d8fa60>, parameters={'amount': {'type': 'float', 'description': '待转换的金额'}, 'from_currency': {'type': 'str', 'description': '源货币代码(如 USD)'}, 'to_currency': {'type': 'str', 'description': '目标货币代码(如 EUR)'}})
非常棒!我们成功提取了将提供给 LLM 作为工具定义的信息。
设计系统提示
我们将使用 gpt-4o-mini 作为推理引擎。已知 GPT 模型系列在输入提示以 JSON 格式时表现更佳,因此我们也将这样做。系统提示是智能体最重要的部分,以下是我们最终使用的系统提示(为简洁起见,部分内容以概述形式呈现):
{ "role": "AI Assistant", "capabilities": [ "在必要时使用提供的工具帮助用户", "对于不需要使用工具的问题直接回复", "规划高效的工具使用顺序" ], "instructions": [ "仅在任务需要时使用工具", "如果问题可以直接回答,则用简单信息回复而非使用工具", "当需要工具时,高效规划使用以减少工具调用次数" ], "tools": [ // 工具列表,包含名称、描述和参数信息 ], "response_format": { "type": "json", "schema": { // 定义响应格式,包括是否需要工具、直接回答、推理过程、计划步骤和工具调用等字段 }, "examples": [ // 示例1:货币转换,需要工具 // 示例2:货币转换,需要工具 // 示例3:直接回答,无需工具 ] }}
我们逐部分分析:
- 角色与能力:定义智能体的角色为“AI 助手”,并说明其能力,包括在必要时使用工具、直接回答问题以及规划工具使用顺序。指令:明确指示智能体仅在必要时使用工具,如果可以直接回答则避免使用工具,并在需要工具时高效规划。工具列表:将工具信息(名称、描述、参数)以 JSON 格式提供给系统提示。响应格式:定义 LLM 的输出格式为 JSON,确保包含是否需要工具、直接回答、推理过程、计划步骤和工具调用等信息。示例:提供多个示例,展示工具使用和直接回答的场景,帮助 LLM 理解预期行为。
实现智能体类
智能体类的代码较长,主要由于系统提示内容较多。以下为核心逻辑概述,完整代码请参考 GitHub 仓库。智能体类包含以下方法:初始化、添加工具、使用工具、创建系统提示、规划和执行操作。每个方法的功能如下:
class Agent: def __init__(self): """初始化智能体,工具注册表为空。""" self.client = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY")) self.tools: Dict[str, Tool] = {} def add_tool(self, tool: Tool) -> None: """向智能体注册新工具。""" self.tools[tool.name] = tool def use_tool(self, tool_name: str, **kwargs: Any) -> str: """使用指定参数执行特定工具。""" if tool_name not in self.tools: raise ValueError(f"工具 '{tool_name}' 未找到。可用工具:{list(self.tools.keys())}") tool = self.tools[tool_name] return tool.func(**kwargs) def create_system_prompt(self) -> str: """为 LLM 创建包含可用工具的系统提示。""" # 返回格式化的系统提示,包含角色、能力、指令、工具列表及响应格式 def plan(self, user_query: str) -> Dict: """使用 LLM 为工具使用制定计划。""" # 调用 LLM 生成计划,返回 JSON 格式的响应 def execute(self, user_query: str) -> str: """执行完整流程:规划并执行工具。""" plan = self.plan(user_query) if not plan.get("requires_tools", True): return plan["direct_response"] results = [] for tool_call in plan["tool_calls"]: tool_name = tool_call["tool"] tool_args = tool_call["args"] result = self.use_tool(tool_name, **tool_args) results.append(result) return f"思考:{plan['thought']}\n计划:{'. '.join(plan['plan'])}\n结果:{'. '.join(results)}"
execute
方法首先调用 plan
方法生成计划。如果计划中不需要工具,则直接返回直接回答。如果需要工具,则按计划顺序执行工具,并将结果组合返回。
运行智能体
我们已经完成了创建和使用智能体的所有必要代码。以下代码初始化智能体,添加货币转换工具,并处理两个用户查询。第一个查询需要使用工具,第二个不需要:
agent = Agent()agent.add_tool(convert_currency)query_list = ["我从塞尔维亚去日本旅行,带了 1500 本地货币,能换多少日元?", "你好吗?"]for query in query_list: print(f"\n查询:{query}") result = agent.execute(query) print(result)
预期输出类似于:
查询:我从塞尔维亚去日本旅行,带了 1500 本地货币,能换多少日元?思考:我需要使用货币转换工具将 1500 塞尔维亚第纳尔(RSD)转换为日元(JPY)。计划:使用 convert_currency 工具将 1500 RSD 转换为 JPY。返回转换结果。结果:1500 RSD = 2087.49 JPY查询:你好吗?我只是一个计算机程序,没有感情,但我在这里,随时准备帮助你!
正如预期,第一个查询使用了工具,而第二个查询直接给出了回答。
今日总结
今天我们学习了:
- 如何将 Python 函数包装为工具提供给智能体;如何设计系统提示,利用工具定义规划执行流程;如何实现一个智能体,执行规划中的操作。
希望本文能激发您对 AI 智能体的兴趣,并鼓励您在自己的项目中尝试这些技术。