释放 Ai 的力量：建立自己的数据分析师 Ai 代理的分步指南》！
- Rifx.Online
- Large Language Models , Data Science , AI Applications
- 23 Feb, 2025
在2025年开放数据科学大会人工智能构建者峰会上，代理人工智能和人工智能代理是数据社区广泛讨论的话题。如果您是一名数据分析师、数据科学家或数据工程师，您会在专业网络中每天看到提到代理人工智能的多种架构、框架、技术和用例。
在这种信息丰富且主题复杂的情况下，从最基本的实现开始建立对该主题的理解是值得的，这也是本文及其附带的笔记本的目的。
数据分析师人工智能代理的基本架构图
代理人工智能的简单术语
代理、智能代理和自主代理 是在商业大型语言模型（LLMs）广泛可用之前，学术界就已讨论的概念。简单来说，代理是一个系统，当提供一个目标和一组工具时，它会通过使用提供的工具来实现该目标。
使用大型语言模型（LLMs）来确定实现目标的路径、工具使用的顺序以及管理意外状态通常被称为代理人工智能。
与基于规则的方法或其他机器学习替代方案相比，LLMs的两个关键特性使其作为代理的编排引擎特别有效：
嵌入知识
LLMs包含从其训练数据中提取的大量信息。这使它们能够在最少的指令下管理各种状态。
结构化响应
LLMs可以被指示生成结构化输出。这简化了它们在代理系统中的实现，代理系统的核心是处理这些输出的程序。
我们的数据驱动人工智能代理
在我们的例子中，我们构建了一个具有以下特征的人工智能代理：
目标
根据数据库中包含的信息，用英语回答用户提出的与业务相关的问题。
工具
- 一个在 Teradata Vantage 数据库中执行结构化查询语言语句的函数。
需求
要构建这个代理，我们需要以下资源：
- 一个集成了 Jupyter Notebook 的 Teradata VantageCloud 开发环境。在 ClearScape 分析体验中免费获取一个。
- 对大型语言模型 (LLM) 的 API 访问。
在示例笔记本和本文中，我们使用 OpenAI，但这可以更改。这样做需要相应地修改 API 调用。
准备开发环境
- 登录到 ClearScape 分析体验。
- 在 ClearScape 分析体验控制台中创建一个环境。
请记下您选择的密码，因为您将需要它与数据库进行交互。
- 通过点击“运行演示”来启动 Jupyter Notebook 环境。
-
在 Jupyter Notebook 环境中，打开“用例”文件夹，并创建一个您选择的名称的文件夹。
-
打开您创建的文件夹，在“文件”菜单下点击“从 URL 打开”，并在对话框中粘贴以下 URL：
- 这将把项目的笔记本加载到您的环境中。
-
在您创建的文件夹中，创建或加载一个
configs.json文件，结构如下，将"your-api-key-here"替换为您的实际 LLM API 密钥：{ "llm-api-key": "your-api-key-here" } -
依赖项说明
- 如果您在 ClearScape 分析体验上运行该项目，上述都是所需的前提条件；该笔记本可以直接运行。
- 如果您使用不同的 LLM 提供商，您需要安装相应的 SDK，并相应调整笔记本中的代码。
构建数据分析师人工智能代理
项目设置
在本节中，我们将安装和导入必要的库，将大型语言模型服务 API 密钥加载到内存中，创建数据库连接，并加载示例数据。
第一步是安装我们选择的 LLM API 提供商的 SDK。
在 ClearScape 分析体验中运行项目显著简化了此设置，因为数据库已集成在笔记本环境中。创建与该数据库的连接非常简单，如下片段所示。
%run -i ../startup.ipynb
eng = create_context(host='host.docker.internal', username='demo_user', password=password)
print(eng)
一旦创建了数据库连接，就可以对数据库运行 data_loading_queries 以创建示例所需的表。
代理配置
代理的配置由两个部分组成：
- 代理日常工作的定义。
- 可供代理实现目标的工具定义。
代理的日常工作通常在系统提示中定义，概述其目标和实现该目标所需的必要操作。
system_prompt = f"""
You are a data analyst for a retail company working with a Teradata system.
1. Users send you business questions in plain English, and you provide answers to those questions based on the data in the provided databases.
2. To generate answers, you must construct an SQL statement to query the Teradata database. The catalog of databases, tables, and columns available for querying is provided in the following JSON structure: {query_teradata_dictionary(databases)}.
- The SQL query must be written as a single line of text without carriage returns or line breaks.
- Ensure that the query adheres to Teradata's SQL dialect and does not include unsupported keywords such as `LIMIT`.
- Joins across tables should be used when necessary to fulfill the user's request.
3. Execute the SQL query in Teradata.
4. Present the query results to the user in plain English.
"""
为了我们的目的，系统提示应嵌入满足用户请求所需的数据目录。
函数 query_teradata_dictionary() 负责检索数据目录。Teradata 系统中的 DBC 数据库包含可以轻松检索的相关信息，以形成数据目录。
databases = ["teddy_retailers"]
def query_teradata_dictionary(databases_of_interest):
query = f'''
SELECT DatabaseName, TableName, ColumnName, ColumnFormat, ColumnType
FROM DBC.ColumnsV
WHERE DatabaseName IN ('{', '.join(databases_of_interest)}')
'''
table_dictionary = DataFrame.from_query(query)
return json.dumps(table_dictionary.to_pandas().to_json())
可供代理使用的工具被定义为函数。这些函数的知识通过称为函数签名的数据结构提供给大型语言模型，使其能够生成适当的函数调用。
我们只需为代理提供一个工具，即查询数据库所需的函数。
def query_teradata_database(sql_statement):
query_statement = sql_statement.split('ORDER BY', 1)[0]
query_result = DataFrame.from_query(query_statement)
return json.dumps(query_result.to_pandas().to_json())
由于我们需要以 JSON 结构的方式将查询结果返回给代理，因此使用 Teradata 数据框处理 SQL 查询非常方便。然而，这种方法排除了在要处理的 SQL 查询中使用 ORDER BY 语句。大型语言模型对此并不知情，需要通过提示测试和反复试验使其生成与该限制兼容的 SQL 查询。
提示不会提供一致的结果，规则会。当使用大型语言模型时，如果某些内容可以制定为规则，最好将其制定为规则，因此在函数开头进行的小字符串操作。
如前所述，工具通过其相应的签名为大型语言模型所知。函数签名可以手动创建或使用辅助函数生成。后者是一种比手动定义更具可扩展性和更少错误的方式。
def function_to_schema(func) -> dict:
type_map = {
str: "string",
int: "integer",
float: "number",
bool: "boolean",
list: "array",
dict: "object",
type(None): "null",
}
try:
signature = inspect.signature(func)
except ValueError as e:
raise ValueError(
f"Failed to get signature for function {func.__name__}: {str(e)}"
)
parameters = {}
for param in signature.parameters.values():
try:
param_type = type_map.get(param.annotation, "string")
except KeyError as e:
raise KeyError(
f"Unknown type annotation {param.annotation} for parameter {param.name}: {str(e)}"
)
parameters[param.name] = {"type": param_type}
required = [
param.name
for param in signature.parameters.values()
if param.default == inspect._empty
]
return {
"type": "function",
"function": {
"name": func.__name__,
"description": (func.__doc__ or "").strip(),
"parameters": {
"type": "object",
"properties": parameters,
"required": required,
},
},
}
代理运行时
对于代理运行时，我们需要以下内容：
-
提取我们工具函数的数据结构。
tools = [query_teradata_database] tool_schemas = [function_to_schema(tool) for tool in tools] -
创建一个映射，将我们函数的名称（字符串）与 Python 中的实际函数对象关联起来。大型语言模型返回代理需要执行的函数名称作为字符串，但我们在代码中需要调用的是函数对象。这个映射将大型语言模型的响应与我们的代理运行时连接起来。
tools_map = {tool.__name__: tool for tool in tools} -
一个工具调用执行器，这是一个辅助函数，接受函数名称（字符串）和函数参数，并通过我们刚刚创建的 tools_map 执行相应参数的函数。
def execute_tool_call(tool_call, tools_map): name = tool_call.function.name args = json.loads(tool_call.function.arguments) print(f"Assistant: {name}({args})") return tools_map[name](**args) -
我们的运行时函数本身：
该函数接受系统提示和用户提示作为参数。其内部工作包含在两个循环中。外部循环与大型语言模型交互，将大型语言模型的响应附加到消息链中。
def run_full_turn(system_message, messages): while True: print(f"just logging messages {messages}") response = client.chat.completions.create( model="gpt-4o", messages=[{"role": "system", "content": system_message}] + messages, tools=tool_schemas or None, seed=2 ) message = response.choices[0].message messages.append(message) # 内部循环遍历所有需要的工具调用 if message.tool_calls: for tool_call in message.tool_calls: result = execute_tool_call(tool_call, tools_map) result_message = { "role": "tool", "tool_call_id": tool_call.id, "content": result, } messages.append(result_message) else: break
整合所有内容
外循环第一次迭代：
消息链： 此时仅用户查询
[{'role': 'user', 'content': 'what is the month with the greatest amount of orders'}]
LLM 响应：
message = ChatCompletionMessage(content=None, refusal=None, role='assistant', audio=None, function_call=None,
tool_calls=[ChatCompletionMessageToolCall(id='call_OokTYjelsTCfERs9pTfUwA9h',
function=Function(arguments='{"sql_statement":"SELECT EXTRACT(MONTH FROM order_date) AS order_month, COUNT(order_id) AS order_count FROM teddy_retailers.source_orders GROUP BY order_month ORDER BY order_count DESC"}'
, name='query_teradata_database'), type='function')])
LLM 已识别工具并生成将要执行的 SQL 语句。该响应被附加到消息链中。
更新的消息链：
[{'role': 'user', 'content': 'what is the month with the greatest amount of orders'},
ChatCompletionMessage(content=None, refusal=None, role='assistant', audio=None,
function_call=None,
tool_calls=[ChatCompletionMessageToolCall(id='call_OokTYjelsTCfERs9pTfUwA9h',
function=Function(arguments='{"sql_statement":"SELECT EXTRACT(MONTH FROM order_date) AS order_month, COUNT(order_id) AS order_count FROM teddy_retailers.source_orders GROUP BY order_month ORDER BY order_count DESC"}'
, name='query_teradata_database'), type='function')]
这一部分很重要，因为消息链为 LLM 提供了有关工具及其结果的上下文。这就是每个 tool_call 具有 id 的原因。
ChatCompletionMessageToolCall：
ChatCompletionMessageToolCall(id='call_OokTYjelsTCfERs9pTfUwA9h'...)
代理系统执行工具调用并根据内部 for 循环中定义的逻辑将结果附加到消息链中。
更新的消息链，这次包含数据库查询的结果：
[{'role': 'user', 'content': 'what is the month with the greatest amount of orders'},
ChatCompletionMessage(content=None, refusal=None, role='assistant', audio=None,
function_call=None,
tool_calls=[ChatCompletionMessageToolCall(id='call_OokTYjelsTCfERs9pTfUwA9h',
function=Function(arguments='{"sql_statement":"SELECT EXTRACT(MONTH FROM order_date) AS order_month, COUNT(order_id) AS order_count FROM teddy_retailers.source_orders GROUP BY order_month ORDER BY order_count DESC"}',
name='query_teradata_database'), type='function')]),
{'role': 'tool', 'tool_call_id': 'call_OokTYjelsTCfERs9pTfUwA9h',
'content':
'"{\\\\"order_month\\\\":{\\\\"0\\\\":11,\\\\"1\\\\":8,\\\\"2\\\\":12,\\\\"3\\\\":7,\\\\"4\\\\":10,\\\\"5\\\\":9},\\\\
"order_count\\\\":{\\\\"0\\\\":607,\\\\"1\\\\":810,\\\\"2\\\\":721,\\\\"3\\\\":623,\\\\"4\\\\":780,\\\\"5\\\\":902}}"'}
外循环第二次迭代：
消息链： 同上
[{'role': 'user', 'content': 'what is the month with the greatest amount of orders'},
ChatCompletionMessage(content=None, refusal=None, role='assistant', audio=None,
function_call=None,
tool_calls=[ChatCompletionMessageToolCall(id='call_OokTYjelsTCfERs9pTfUwA9h',
function=Function(arguments='{"sql_statement":"SELECT EXTRACT(MONTH FROM order_date) AS order_month, COUNT(order_id) AS order_count FROM teddy_retailers.source_orders GROUP BY order_month ORDER BY order_count DESC"}',
name='query_teradata_database'), type='function')]),
{'role': 'tool', 'tool_call_id': 'call_OokTYjelsTCfERs9pTfUwA9h',
'content':
'"{\\\\"order_month\\\\":{\\\\"0\\\\":11,\\\\"1\\\\":8,\\\\"2\\\\":12,\\\\"3\\\\":7,\\\\"4\\\\":10,\\\\"5\\\\":9},\\\\
"order_count\\\\":{\\\\"0\\\\":607,\\\\"1\\\\":810,\\\\"2\\\\":721,\\\\"3\\\\":623,\\\\"4\\\\":780,\\\\"5\\\\":902}}"'}
LLM 响应：
message = ChatCompletionMessage(
content='The month with the greatest number of orders is September, with a total of 902 orders.',
refusal=None, role='assistant', audio=None, function_call=None, tool_calls=None)
此时，LLM 使用最终的消息链提供最终响应。
结论
如在此示例项目中所见，代理人工智能根本上是关于编排的。人工智能代理不是大型语言模型；相反，它们是集成了一组工具、规则和保护措施的系统。在大型语言模型的编排下，这些组件协同工作以实现特定目标。
大型语言模型无法执行外部工具；代理系统根据大型语言模型提供的响应执行工具。
大型语言模型提供商，如 OpenAI，提供具有定义数据结构的 API，旨在以下目的：
- 参考工具并定义工具调用
- 组织消息链
- 记录工具响应
在用户自己的基础设施上运行的大型语言模型可能无法提供所有这些功能，因此在这种情况下，代理框架变得更加必要。
通过为工具及其输出引入错误处理机制，可以改进此示例。
