本文旨在回答“MARS如何通过无缝融合结构化和非结构化数据处理来彻底改变企业AI”这个问题，通过提供一个快速的回顾，并解释入门代码，以帮助任何对构建基于MARS的企业AI应用感兴趣的人。

对于其他人，让我们来看看我们要构建什么。我们的示例是一个简单的工作流，它使用OpenAI的Swarm库来协调一个智能体，该智能体向数据库（SingleStore）发出查询，并使用Nvidia的Nemo guardrails，这是一个强大的库，用于执行输入/输出验证，并以易于使用、可配置的方式提供查询保护。

你可能会问，为什么要用智能体来查询数据库？

因为许多企业都拥有结构化和非结构化数据，而我们想证明你可以在一个SQL语句中运行选择、分组、聚合、向量搜索和精确关键字匹配。考虑到我们讨论的是企业应用，我们不能仅仅依赖文本到SQL的生成模型，因为在大型企业中，存在大量SQL语句（往往长达数页），而大型语言模型（LLM）目前还无法确定且准确地生成这些语句。

但首先，让我们从基础开始。

什么是智能体？

在其最简单的形式中，我们可以将智能体视为一种程序，它能够根据其环境和输入做出决策并采取行动。我们可以通过想象一个实体来理解这个概念，这个实体与人类类似，可以访问三个关键要素：1）智力（通过LLM），2）工具（例如，通过API调用其他应用程序、浏览互联网等的能力），以及可选的3）对特定知识的访问（包括结构化和非结构化数据）。

为了更好地理解这一点，让我们逐一深入了解这些关键组件：

1. 智力——带有系统提示的LLM

智能体通常从一个LLM开始，该LLM通常配备了一个系统提示，为其提供了操作指令和特定的角色。系统提示奠定了基础——决定了智能体应该是信息性的、说服性的，还是甚至是娱乐性的。这允许开发人员塑造智能体与用户交互的语气和风格。为了使智能体更加专业化，不同的智能体可以访问不同的微调LLM。

2. 工具——函数和API访问

LLM很强大，但如果不将它们与现实世界连接起来，它们就只是花哨的聊天机器人。工具为智能体提供了功能，使其智力能够超越响应查询，执行动作，因此称为“代理性”。这些工具可以包括从调用API、浏览互联网到自定义函数（用于从数据库中检索数据，如我们的示例）的任何内容。将工具视为智能体的四肢，使其能够与外部环境交互，从而使其更具行动力。

3. 知识（可选）

智能体还可以配备知识——换句话说，就是高度特定的系统提示以及使其在某些领域有效的专门数据。这些知识可以采用几种不同的形式：

a) 非结构化数据

非结构化数据，如PDF、Markdown文件等，可用于丰富智能体的响应。一个恰当的例子是OpenAI的助手，它们可以访问来自文档、网页或PDF的知识，这些知识首先用于回答问题，而不是依赖LLM的知识。

b) 记忆/会话以跟踪历史

记忆允许智能体跟踪用户交互，并在会话中记住关键细节。例如，智能体可以回忆起用户喜欢动作电影，或者记住过去问题中的细节，从而使交互更加顺畅和个性化。

让我们构建一个简单的智能体

为了将所有这些付诸实践，让我们构建一个简单的智能体，它根据用户偏好推荐电影，通过查询数据库来实现。这个智能体将利用SingleStore，这是一个数据库，允许你检索SQL数据、运行分析查询，甚至使用向量和关键字搜索——所有这些都在一个地方完成。

为什么选择SingleStore？

SingleStore是一个自然的选择，因为它将多种功能组合到一个SQL查询中。无需拼凑不同的数据库、搜索引擎和分析解决方案，SingleStore可以处理所有这些——包括高级的向量和关键字搜索。这使得它非常适合我们的电影推荐智能体。

接下来，对于任何企业应用来说，安全性和合规性都是关键考量因素。这就是为什么我们还集成了保护机制来提供输入和输出验证，确保安全的用户体验。如上所述，我们将使用Nvidia强大的Nemo保护机制来实现这一点，因为它提供了稳健的输入/输出验证，并且特别设计为与LLM无缝协作，与其他替代方案相比，确保了高水平的可靠性和合规性。

实施步骤

我们将采取以下步骤来构建我们的电影推荐代理：

1. 我们安装Nemo guardrails，并将我们的策略/规则放入一个名为‘rails.co’的文件中，该文件位于一个名为nemo-configs的文件夹中（你可以将文件夹命名为任何名称，只要在实例化防护栏时正确加载它即可）。
2. 然后，我们创建几个使用singlestoredb库连接到SingleStore并执行查询的辅助函数。下面的示例使用了一个非常简单的select语句，但你可以根据自己的数据库架构和数据来定制它。我们将这个函数称为search_movies。
3. 接下来，我们创建一个代理，并为其提供名称、指令（系统提示），最后让它可以访问我们的函数——search_movies。在这个例子中，我们没有为代理指定不同的模型，但你可以通过添加一个名为model=“your_model_name”的参数来添加自己的模型。
4. 最后，我们将所有这些整合在一起，以便当查询进来时，首先将其传递给防护栏。如果防护栏认为输入是可接受的（通过使用我们的rails.co文件），那么它会根据查询的适当性来决定是直接回应而不采取进一步行动，还是调用SingleStore代理进行电影推荐。

就这样，简单明了。

你现在可以使用这个方法来创建其他代理，并将它们不断添加到你的Swarm对象中。然而，请注意，随着代理数量的增加，管理它们的交互并确保最佳性能可能会变得具有挑战性。务必监控系统是否存在延迟问题和潜在的冲突，并考虑采用负载均衡或模块化架构等策略来解决这些挑战。换句话说，这个例子同样适用于软件工程的原则。

此应用的代码

以下是我们的电影推荐代理的完整实现：

python
from swarm import Swarm, Agent
import singlestoredb as s2
import os
from nemoguardrails import LLMRails, RailsConfig
from openai import OpenAI
from typing import Dict, Any, List, Callable, Tuple
import numpy as np
# Initialize OpenAI
client = OpenAI(api_key=os.getenv(“OPENAI_API_KEY”))
# Initialize NeMo Guardrails
config = RailsConfig.from_path(“nemo-configs/”)
rails = LLMRails(config)
# Global connection and query
singlestore_conn = None
current_query = “”
def connect_to_singlestore():
 “Establish connection to SingleStore”
 try:
 # Get connection parameters from environment
 host = os.getenv(“SINGLESTORE_HOST”)
 user = os.getenv(“SINGLESTORE_USER”)
 password = os.getenv(“SINGLESTORE_PASSWORD”)
 database = os.getenv(“SINGLESTORE_DATABASE”
# Connect using the working format
 return s2.connect(host=host,
 port=3306,
 user=user,
 password=password,
 database=database)
 except Exception as e:
 print(f”Error connecting to SingleStore: {e}”)
 return None
def search_movies() -> str:
 "Core function to search movies in SingleStore"
 global singlestore_conn, current_query
 
 try:
 if not singlestore_conn:
 singlestore_conn = connect_to_singlestore()
 if not singlestore_conn:
 return “Failed to connect to database”
sql_query = ```
 SELECT title, MATCH(title) AGAINST (?) as relevance
 FROM movies
 WHERE MATCH(title) AGAINST (?)
 ORDER BY relevance DESC
 LIMIT 10
 ```
cursor = singlestore_conn.cursor()
 cursor.execute(sql_query, (current_query, current_query))
 
 # Convert cursor results to numpy array for easier handling
 results = np.array(cursor.fetchall())
 cursor.close()
if len(results) == 0:
 return “No movie recommendations found for your query.”
 
 response = “Here are some movie recommendations:\n”
 for title, relevance in results:
 response += f”- {title} (Relevance: {float(relevance):.2f})\n”
 
 return response
 
 except Exception as e:
 if singlestore_conn:
 singlestore_conn.close()
 singlestore_conn = None
 return f”Error getting recommendations: {e}”
def direct_llm_response(query: str) -> str:
 “””Get response directly from LLM for non-SingleStore queries”””
 try:
 response = client.chat.completions.create(
 model=”gpt-3.5-turbo”,
 messages=[
 {“role”: “user”, “content”: query}
 ]
 )
 return response.choices[0].message.content or “No response generated”
 except Exception as e:
 return f”Error getting LLM response: {e}”
def is_singlestore_query(response) -> bool:
 “””Check if the guardrails response indicates need for SingleStore”””
 try:
 # Check if response contains specific indicators from our rails.co rules
 response_text = response.last_message.content.lower()
 return “inform using singlestore” in response_text or \
 “delegate to agent” in response_text
 except AttributeError:
 # If we can’t access the response content as expected,
 # default to treating it as a non-SingleStore query
 return False
def main():
 global current_query
 
 # Initialize Swarm client
 swarm_client = Swarm()
 
 # Initialize agent with the movie recommendation function
 agent = Agent(
 name=”MovieRecommendationAgent”,
 instructions=”You are a helpful movie recommendation agent.”,
 functions=[search_movies]
 )
print(“Welcome! You can ask me anything. Type ‘exit’ to quit.”)
 
 while True:
 # Get user input
 user_query = input(“\nYou: “).strip()
 
 if user_query.lower() == ‘exit’:
 print(“Goodbye!”)
 if singlestore_conn:
 singlestore_conn.close()
 break
 
 try:
 # First pass through guardrails
 guardrails_response = rails.generate(messages=[{“role”: “user”, “content”: user_query}])
 
 # Check if query needs SingleStore
 if is_singlestore_query(guardrails_response):
 # Update current query for the search function
 current_query = user_query
 
 # Use SingleStore agent for movie recommendations
 messages = [{“role”: “user”, “content”: user_query}]
 response = swarm_client.run(agent=agent, messages=messages)
 print(“Bot:”, response.messages[-1][“content”])
 else:
 # Use direct LLM response for general queries
 response = direct_llm_response(user_query)
 print(“Bot:”, response)
 
 except Exception as e:
 print(f”An error occurred: {e}”)
if __name__ == “__main__”:
 main()

结论

在本文中，我们看到了一个简单的示例，展示了如何通过将 OpenAI 的功能与 SingleStore 的多功能数据库功能相结合并使用 NemoGuardrails 添加一层安全性来构建基于 MARS 的企业 AI 应用程序。