MindsDB 销售分析智能体实战：一条 SQL 打通 MongoDB 与 HubSpot，并用 AI Agent 回答业务问题-创锋一号

MindsDB 销售分析智能体实战：一条 SQL 打通 MongoDB 与 HubSpot，并用 AI Agent 回答业务问题

【免费下载链接】ai-engineering-hubIn-depth tutorials on LLMs, RAGs and real-world AI agent applications.项目地址: https://gitcode.com/GitHub_Trending/ai/ai-engineering-hub

导读

本指南基于开源仓库ai-engineering-hub中的sales-analytics-agent项目，完整演示如何将MongoDB（产品评价 reviews 与销售明细 product_sales）和HubSpot（企业 companies 与联系人 contacts）接入MindsDB，先通过单库与跨库 SQL 完成销售分析，再基于同一份数据创建sales_agent，让它用自然语言回答"谁是营收最高的客户""某客户处于什么生命周期阶段、累计消费多少"这类业务问题。读完本文，你将掌握 MindsDB 数据源接入、跨数据库 JOIN、CREATE AGENT智能体配置以及 Python SDK 流式查询的完整实战链路。

一、项目概览：为什么需要"数据库 + Agent"的组合

传统的销售分析流程通常分两步走：分析师用 SQL 写查询，再人工把结果翻译成业务结论；而sales-analytics-agent项目展示了一条更直接的路径——把 CRM（HubSpot）与交易流水（MongoDB）作为 MindsDB 的虚拟数据库，既能用标准 SQL 做精确统计，又能让 AI Agent 直接"看懂"表结构并自动生成查询、汇总答案。

核心步骤记录在 QUERY.md 中，共五步：创建数据库 → 显示数据 → 查询数据 → 创建 Agent → 查询 Agent；配套的 main.py 展示了使用 MindsDB Python SDK 流式获取 Agent 回答的实现；README.md 则给出环境搭建与运行说明。整个项目是ai-engineering-hub仓库（93+ 个可运行项目）中面向"数据层 + AI 层"的中级实战示例，适合已经熟悉 SQL、希望让大模型直接驾驭业务数据的开发者。

二、环境准备：启动 MindsDB

2.1 需要的凭证

服务	用途	使用阶段
HubSpotaccess token	创建`hubspot_data`数据库，读取 companies / contacts	`CREATE DATABASE`
OpenAIAPI key	供 AI Agent 调用 LLM 生成 SQL 与回答	`CREATE AGENT`
MongoDB连接串	创建`mongodb_data`数据库	`CREATE DATABASE`

说明：项目作者在示例中已经预置了 HubSpot 与 MongoDB 的数据源（见 QUERY.md 中的 demo 连接串），可以直接沿用；若使用自己的数据，替换为自有凭证即可。

2.2 启动 MindsDB 容器

MindsDB 以 Docker 镜像方式运行，同时暴露两个端口：

47334：HTTP API（REST），也是 Python SDK 默认连接端口；
47335：MySQL 协议端口，任何 MySQL 客户端都可以连上来执行 SQL。

docker run --name mindsdb_container \ -e MINDSDB_APIS=http,mysql \ -p 47334:47334 -p 47335:47335 \ mindsdb/mindsdb:latest

MINDSDB_APIS=http,mysql显式声明只启用 HTTP 与 MySQL 两种 API，控制面更干净。容器启动后，MindsDB 会自动创建内置的mindsdb数据库，后续的 Agent 就注册在这里。

2.3 安装 Python SDK（可选）

如果希望用 Python 而非 SQL 查询 Agent，需要 Python 3.8+ 与 SDK：

pip install mindsdb_sdk

三、第一步：把 MongoDB 与 HubSpot 创建为 MindsDB 数据源

MindsDB 的核心能力之一是将外部数据库注册为虚拟数据库，之后即可用统一的 SQL 方言访问。这一步在 QUERY.md 中通过两条CREATE DATABASE语句完成。

MongoDB（demo 连接串，包含 reviews 与 product_sales 两个集合）：

CREATE DATABASE mongodb_data WITH ENGINE = 'mongodb', PARAMETERS = { "host": "mongodb+srv://demouser:MindsDB_demo@mindsdb-demo.whljnvh.mongodb.net/demo" };

HubSpot（使用 access token 认证）：

CREATE DATABASE hubspot_data WITH ENGINE = 'hubspot', PARAMETERS = { "access_token": "pat-na1-d7a93c10-07d1-45fd-ac7a-b109d00d9e2b" };

语法要点：WITH ENGINE声明数据源类型（MindsDB 支持数十种引擎），PARAMETERS传入该引擎所需的连接参数。MongoDB 使用标准mongodb+srv://连接串；HubSpot 则通过 Private App 生成的pat-na1-*形式 access token 认证。创建成功后，表名即映射为数据库名.表名的三段式引用，例如hubspot_data.companies、mongodb_data.reviews。

四、第二步：验证数据可见性

建库后先用LIMIT 10抽查各表，确认字段名与数据类型符合预期，这是后续写 JOIN 与 Agent 提示词的前提。

HubSpot 企业表（抽样字段：公司名、域名、行业、年营收、员工数）：

SELECT name, domain, industry, annualrevenue, numberofemployees FROM hubspot_data.companies LIMIT 10;

MongoDB 评价表（抽样字段：客户邮箱、SKU、星级、标题、正文）：

SELECT customer_email, product_sku, star_rating, review_title, review_text FROM mongodb_data.reviews LIMIT 10;

五、第三步：单库与跨库 SQL 查询

5.1 单库查询

Q. 产品评价的平均星级是多少？

SELECT AVG(star_rating) AS average_rating FROM mongodb_data.reviews;

AVG(...) AS average_rating为聚合结果起别名，便于下游程序解析。

5.2 跨库查询（核心能力）

Q. 按营收排名的 Top 10 客户是谁？

SELECT hc.firstname, hc.lastname, ROUND(SUM(ps.sales_amount), 2) AS total_revenue FROM mongodb_data.product_sales AS ps JOIN hubspot_data.contacts AS hc ON ps.customer_email = hc.email GROUP BY hc.firstname, hc.lastname ORDER BY total_revenue DESC LIMIT 10;

这是整个示例的"技术题眼"：product_sales来自MongoDB，contacts来自HubSpot，而 MindsDB 允许在一条 SQL 里跨两个物理数据库做JOIN。连接键是product_sales.customer_email = contacts.email——客户的交易邮箱与 CRM 联系人邮箱一致，这个关系在第六节创建 Agent 时还会显式告知模型。ROUND(SUM(...), 2)保证金额保留两位小数。

六、第四步：创建销售分析 AI Agent

跨库 JOIN 解决了"如何查"，而CREATE AGENT解决"如何让业务人员直接问"。

CREATE AGENT sales_agent USING model = { "provider": "openai", "model_name": "gpt-5.4", "api_key": "<your-openai-api-key>" }, data = { "tables": [ "hubspot_data.contacts", "mongodb_data.product_sales" ] }, prompt_template = ' hubspot_data.contacts stores CRM contact data including: id, email, firstname, lastname, jobtitle, company, city, website, lifecyclestage, hs_lead_status. mongodb_data.product_sales stores sales order line items including: order_number, order_date, customer_email, city, state, product_sku, sales_quantity, discount_percent, sales_price, sales_amount, sales_tax_percent, tax_amount. To join both sources, use: hubspot_data.contacts.email = mongodb_data.product_sales.customer_email. ', timeout = 60;

逐参数拆解：

model：指定 Agent 背后的大模型。provider: "openai"+model_name指明模型系列，api_key传入凭证。示例使用gpt-5.4，实际替换为你可用的模型标识即可。
data.tables：声明 Agent 可访问的表清单。这里只开放hubspot_data.contacts与mongodb_data.product_sales两张表——最小权限原则：不把全部表暴露给模型，避免它"看到"不该用的数据，也缩小了 SQL 生成空间、提升准确率。
prompt_template：这是 Agent 质量的胜负手。模板不是废话描述，而是精确给出两张表的字段清单与 JOIN 关系：
- contacts 表：id, email, firstname, lastname, jobtitle, company, city, website, lifecyclestage, hs_lead_status；
- product_sales 表：order_number, order_date, customer_email, city, state, product_sku, sales_quantity, discount_percent, sales_price, sales_amount, sales_tax_percent, tax_amount；
- 并明确连接键contacts.email = product_sales.customer_email。
把 schema 写进提示词后，模型无需猜测字段名，生成的 SQL 可直接执行；把 join 规则写进提示词后，模型才知道如何跨库关联两张表。这是从"能跑"到"跑得准"的关键。
timeout = 60：Agent 单次推理的超时上限（秒），防止长查询挂死。

七、第五步：用 SQL 查询 Agent

Agent 创建后，它本身就是一个虚拟表：传入question，返回answer。

单库问题：

SELECT answer FROM sales_agent WHERE question = 'What is the total revenue in product sales?';

跨库问题：

SELECT answer FROM sales_agent WHERE question = 'What is the lifecycle stage of noah.johnson@clearwaterholdings.co, and how much have they spent in total?';

第二个问题同时涉及两库：lifecycle stage字段在 HubSpot 的 contacts 表，消费金额要汇总 MongoDB 的 product_sales 表——Agent 需要自行完成"定位表 → 生成 JOIN SQL → 执行 → 组织自然语言回答"的全链路。提示词中预置的字段清单与 join 规则，正是保证它在这一步不犯错的基础。

八、进阶：用 Python SDK 流式查询 Agent

对需要把 Agent 集成进业务系统的场景，main.py 给出了 Python 侧的标准写法：

import mindsdb_sdk # connects to the default port (47334) on localhost server = mindsdb_sdk.connect("http://127.0.0.1:47334") agent = server.agents.get("sales_agent") # stream the completion completion = agent.completion_stream( [ { "question": "Who is the top customer by revenue? Include their full name, job title, and company from our CRM.", "answer": None, } ] ) # print the completion for chunk in completion: print(chunk, end="", flush=True)

逐行解读：

mindsdb_sdk.connect("http://127.0.0.1:47334")：连接本地 MindsDB 的HTTP API（即 docker 启动时的 47334 端口）；
server.agents.get("sales_agent")：按名称获取第六节创建的 Agent 对象；
agent.completion_stream([{ "question": ..., "answer": None }])：以对话消息列表形式发起请求，answer置为None表示该轮由模型补全。消息数组结构天然支持多轮上下文扩展；
for chunk in completion：流式（streaming）消费 token，print(chunk, end="", flush=True)实现逐字打印、无需等待完整响应，显著改善长回答的体验，也适合对接 WebSocket/SSE 等实时场景。

注意：此脚本默认认为sales_agent已通过 SQL 创建成功；若尚未创建，会因找不到 Agent 而报错。运行顺序应为"先 SQL 建 Agent，再 Python 查 Agent"。

九、调用链与工程要点小结

从源码与文档可以梳理出完整调用链：

SQL 客户端 / Python SDK │ ▼ MindsDB (Docker: HTTP 47334 / MySQL 47335) │ ├── mongodb_data (MongoDB: reviews, product_sales) └── hubspot_data (HubSpot: companies, contacts) │ ▼ sales_agent (OpenAI 模型 + data.tables + prompt_template)

工程实践要点：

凭证与数据分离：建库凭证写在CREATE DATABASE PARAMETERS，模型凭证写在CREATE AGENT model，均不应提交到版本库；
最小化 Agent 数据面：data.tables只开放必要的表，减少幻觉与越权访问；
schema 前置进提示词：把字段名、类型语义与 JOIN 键写进prompt_template，是 Agent 生成可靠 SQL 的最强杠杆；
先 SQL 后 Agent：先用单库/跨库 SQL 验证数据与关联逻辑（第五节的 Top10 客户查询），再把同样的逻辑交给 Agent 自然语言化，两者可互为校验；
流式接入：生产集成优先用completion_stream，避免长回答阻塞。

十、相关文件与延伸阅读

本文全部实操内容均可在此仓库中直接复现：

项目 README：环境搭建、Docker 启动与运行总览；
分步 SQL 指南 QUERY.md：本文第三至七节所有 SQL 的原始出处；
Python SDK 示例 main.py：流式查询 Agent 的最小实现；
仓库总览 README.md：ai-engineering-hub的整体项目地图，其中 RAG SQL Router、Deploy Agentic RAG 等中级项目与本示例在"数据与 Agent 结合"方向上可相互参照。

适用前提与限制：本示例依赖 MindsDB 对 MongoDB、HubSpot 引擎的内置支持以及外部 OpenAI 服务，model_name与api_key需按实际可用模型替换；示例中的 MongoDB demo 连接串与 HubSpot token 为项目作者提供的演示凭证，生产环境务必换成自有数据源。

【免费下载链接】ai-engineering-hubIn-depth tutorials on LLMs, RAGs and real-world AI agent applications.项目地址: https://gitcode.com/GitHub_Trending/ai/ai-engineering-hub

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

企业官网建设流程全解析