前言

传统查询模式存在几个核心问题：SQL语法学习成本高 -> 复杂查询编写难度大 -> 查询结果缺乏业务语境的解释

MCP协议的引入恰巧能简化这些问题。该协议定义了AI应用与外部数据源之间的标准化通信接口，使得不同系统能够通过统一的方式进行数据交换。

而要真正解决传统查询模式的痛点，仅有协议标准还不够，还需要强大的数据处理引擎和易用智能化的AI框架。这正是Apache Doris和LangChain发挥价值的地方。

Apache Doris作为现代化的实时分析数据库，在性能、扩展性、易用性和生态方面都有着显著优势。其基于MPP架构的向量化执行引擎，能够将查询性能相比传统数据库提升数倍，轻松支持PB级数据的秒级响应。同时，Doris支持在线扩缩容，既能应对业务增长带来的数据量激增，也能满足实时决策的业务需求，大幅降低了运维复杂度，让企业能够专注于业务价值而非技术细节。

在AI应用开发层面，LangChain框架则提供了完整的解决方案。不仅能够构建具备复杂推理能力的智能代理，还拥有丰富的工具生态系统，可以无缝集成各类外部服务。LangChain对主流LLM提供商的兼容支持，确保了最佳的性能和成本平衡，而其内置的优化机制，更是帮助开发者快速构建高质量的AI交互体验。

当Apache Doris遇上LangChain，产生的不仅仅是技术叠加，更是能力的指数级提升。再结合MCP协议，LangChain能够直接理解Doris的表结构和数据特征，自动生成最优查询策略，实现真正的智能化数据访问。

例如当用户输入"哪个客户下单最多"这样的自然语言查询时，Doris MCP + LangChain能够自动进行意图识别、调用对应的 Doris MCP Tool 生成最优的 Doris SQL、执行查询并返回包含深度商业洞察的分析结果。这种无缝且易用的技术融合，实现了"让数据会说话，让洞察触手可及"的愿景。

第一步｜准备环境

fei话到此，接下来直接实测体验下效果如何。

开局我们需要简单准备下基础环境，主要分为Doris、Python和Doris MCP Server环境即可：

1. Doris

如果已经有现成的Doris集群，直接用即可。

若当前还没Doris环境，可以参考Doris官方文档，基于Docker或本地化快速部署搭建一套Doris集群 🔗 ：https://doris.apache.org/zh-CN/docs/dev/gettingStarted/quick-start

实测数据集主要使用 TPC-H Benchmark 🔗：https://doris.apache.org/zh-CN/docs/benchmark/tpch

2. Python环境

Python版本需要>=3.12

主流程coding所需python包（requirements.txt）：

# LangChain MCP 相关依赖
langchain-mcp-adapters
langchain
langchain-community
langchain-openai

# 环境变量管理
python-dotenv

# 异步支持
aiohttp
aiofiles

3. Doris MCP Server

Doris MCP Server 本文荐直接用git clone到本地的方式：

# 1. MCP Server 克隆到本地：
git clone https://github.com/apache/doris-mcp-server.git
cd /本地仓库oris-mcp-server的路径

# 2. 安装依赖：
pip install -r requirements.txt

# 3. 配置数据库信息
cp .env.example .env
vim .env
## Doris FE connection settings
DORIS_HOST=xxx
DORIS_PORT=9030
DORIS_USER=root
DORIS_PASSWORD=xxx
DORIS_DATABASE=xxx

# 4. 启动doris mcp server服务
./start_server.sh &

最后一行出现如下日志 Uvicorn running on 则表示启动成功：

第二步｜手撕代码

经过七七四十九秒的编写调试，完整代码如下：

"""
Doris MCP + LangChain AI Query System
Intelligent database query system with business context analysis.
"""

import asyncio
import logging
from typing import Dict, Any, Optional
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.chat_models import init_chat_model
from langchain.prompts import ChatPromptTemplate
from langchain_mcp_adapters.client import MultiServerMCPClient

# Configuration Constants
DEEPSEEK_API_KEY = "your_deepseek_api_key_here"
MODEL_NAME = "deepseek-chat"
MODEL_PROVIDER = "deepseek"
MCP_CONFIG = {
"doris_mcp_server": {
"transport": "streamable_http",
"url": "http://localhost:3000/mcp"
    }
}
classPromptManager:
"""Business context prompt management for Doris AI assistant."""

    SYSTEM_PROMPT = """
🤖 你是基于Apache Doris数据库的专业AI问数系统，具备强大的数据分析和业务洞察能力。

核心职责：
1. 分析用户问题语境，理解业务需求
2. 精准调用Doris MCP Server工具执行查询
3. 提供深度业务语境分析和专业解释
4. 将技术数据转化为可执行的业务洞察

回答格式要求（必须包含以下结构）：
📈 查询结果：准确的数据查询结果，使用表格或图表符号展示
🔍 业务解读：数据背后的业务含义和趋势分析
💎 关键洞察：重要发现、异常点和数据亮点
🚀 行动建议：基于数据的具体可执行建议

展示风格：
- 使用📊📈📉等图表符号增强数据可视化效果
- 适当使用✅❌⚠️等状态符号标识重要信息
- 用🔥💪⭐等表情符号突出关键发现
- 保持专业、准确、有洞察力的分析风格
- 让数据"说话"，用生动的方式传达专业见解
    """.strip()

    @classmethod
defenhance_query(cls, query: str) -> str:
"""Enhance user query with business context."""
returnf"{cls.SYSTEM_PROMPT}\n\n用户查询：{query}\n请执行查询并提供业务分析，包括数据洞察、业务影响和行动建议。"

classConfig:
"""Application configuration management."""

def__init__(self,
                 api_key: Optional[str] = None,
                 model: Optional[str] = None,
                 provider: Optional[str] = None,
                 mcp_config: Optional[Dict[str, Any]] = None) -> None:

        self.api_key = self._validate_api_key(api_key or DEEPSEEK_API_KEY)
        self.model = model or MODEL_NAME
        self.provider = provider or MODEL_PROVIDER
        self.mcp_config = mcp_config or MCP_CONFIG

    @staticmethod
def_validate_api_key(key: str) -> str:
"""Validate API key configuration."""
ifnot key or key == "your_deepseek_api_key_here":
raise ValueError("Valid DeepSeek API key required")
return key

classDorisMCPAgent:
"""Doris MCP Agent with intelligent business context analysis."""

def__init__(self, config: Config) -> None:
        self.config = config
        self._mcp_client: Optional[MultiServerMCPClient] = None
        self._agent_executor: Optional[AgentExecutor] = None

asyncdefinitialize(self) -> None:
"""Initialize MCP client and LangChain agent."""
        self._mcp_client = MultiServerMCPClient(self.config.mcp_config)

        tools = await self._mcp_client.get_tools()
ifnot tools:
raise RuntimeError("Failed to load tools from MCP server")

        llm = init_chat_model(
            model=self.config.model,
            model_provider=self.config.provider,
            api_key=self.config.api_key
        )

        prompt = ChatPromptTemplate.from_messages([
            ("system", PromptManager.SYSTEM_PROMPT),
            ("human", "{input}"),
            ("placeholder", "{agent_scratchpad}")
        ])

        agent = create_openai_tools_agent(llm, tools, prompt)
        self._agent_executor = AgentExecutor(
            agent=agent,
            tools=tools,
            verbose=True,
            max_iterations=3
        )

asyncdefrun_interactive(self) -> None:
"""Start interactive chat session."""
ifnot self._agent_executor:
raise RuntimeError("Agent not initialized")

        print("\n" + "="*60)
        print("🤖 欢迎使用 Doris AI 问数系统 🤖")
        print("="*60)

        print("🔥 您可以这样问我：")
        print("   1️⃣  当前Doris有哪些库表？")
        print("   2️⃣  请帮我切换到tpch库并分析哪个客户下单最多")
        print("   3️⃣  等等等...")

        print("\n💬 输入您的问题开始分析，输入 'quit' 退出系统")
        print("="*60 + "\n")

whileTrue:
try:
                query = input("You: ").strip()

if query.lower() in {'quit', 'exit', 'q'}:
                    print("Goodbye!")
break

ifnot query:
continue

                enhanced_query = PromptManager.enhance_query(query)
                result = await self._agent_executor.ainvoke({"input": enhanced_query})
                print(f"\nAI: {result['output']}\n")

except KeyboardInterrupt:
                print("\nGoodbye!")
break
except Exception as e:
                print(f"Error: {e}")

asyncdefmain() -> None:
"""Application entry point."""
    logging.basicConfig(level=logging.WARNING)

try:
        config = Config()
        agent = DorisMCPAgent(config)
await agent.initialize()
await agent.run_interactive()
except Exception as e:
        print(f"Error: {e}")
        print("Please check the configuration section")

if __name__ == "__main__":
try:
        asyncio.run(main())
except KeyboardInterrupt:
        print("\nInterrupted")
except Exception as e:
        print(f"Startup failed: {e}")

代码解析

代码经过精简，很多block没有进行过多地细化深入。主要是为了让大家能够快速熟悉Doris MCP + LangChain 搭建AI问数系统的完整流程，后续可以结合自己需求，按模块进行调整应用。

代码主流程如下：

Doris AI 问数系统是一个基于异步架构的智能数据库查询系统，将MCP协议和LangChain框架相结合，为用户提供了一种全新的自然语言数据查询体验。

从系统的执行流程来看，整个过程可以分为三个关键阶段：初始化准备、工具链构建和智能交互。

初始化阶段，通过Config模块验证API密钥等关键配置，创建DorisMCPAgent作为整个系统的管理中心。

接下来进入工具链构建阶段，会建立与Doris数据库的MCP连接，同时初始化DeepSeek聊天模型，并将这些组件整合成一个完整的AgentExecutor执行器。

最后在智能交互阶段，进入持续的用户对话循环，每当用户提出查询时，PromptManager会自动为查询添加专业的业务分析语境，然后LangChain Agent会调用相应的数据库工具执行查询，并生成包含深度商业洞察的分析报告。

整体架构上，异步编程架构确保了系统在处理复杂查询时依然能够保持高效响应，而MCP协议的引入则实现了 AI与Doris数据库的标准化交互，并通过PromptManager的智能增强模块，能够自动为每个查询添加商业分析的专业视角，同时也能结合企业的业务场景进行DIY，使得返回的结果不再是冰冷的数据，而是富有洞察力的商业智能报告。

第三步｜验证效果

代码编写调试并熟悉后，来验证下效果如何👇

首先，直接提问当前Doris有哪些库表？，确认整体流程是否有误：

结果正确，和Doris的库表信息一致，并且给出了数据库表的分类、个别库表的洞察分析和改善建议。

再进行自然语言查询分析测试分析哪个客户下单最多：

结果正确。Doris AI 问数系统自动将自然语言转为了Doris SQL进行查询并返回结果，并且给出了专业的业务解读、关键洞察和行动建议。

怎一个香字了得？

结语

随着MCP协议的进一步完善和AI技术的持续演进，我们有理由相信，AI+数据库的交互模式将成为企业数字化转型的重要推动力。让数据会说话，让洞察触手可及的愿景，正在从理想照进现实。

技术的美妙之处在于化繁为简，而 Doris MCP + LangChain 恰恰做到了这一点。不妨动手试试，或许你会发现，与数据直接对话原来可以如此简单而优雅。

最后

为什么要学AI大模型

当下，⼈⼯智能市场迎来了爆发期，并逐渐进⼊以⼈⼯通⽤智能（AGI）为主导的新时代。企业纷纷官宣“ AI+ ”战略，为新兴技术⼈才创造丰富的就业机会，⼈才缺⼝将达 400 万！

DeepSeek问世以来，生成式AI和大模型技术爆发式增长，让很多岗位重新成了炙手可热的新星，岗位薪资远超很多后端岗位，在程序员中稳居前列。

与此同时AI与各行各业深度融合，飞速发展，成为炙手可热的新风口，企业非常需要了解AI、懂AI、会用AI的员工，纷纷开出高薪招聘AI大模型相关岗位。

最近很多程序员朋友都已经学习或者准备学习 AI 大模型，后台也经常会有小伙伴咨询学习路线和学习资料，我特别拜托北京清华大学学士和美国加州理工学院博士学位的鲁为民老师给大家这里给大家准备了一份涵盖了AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频 全系列的学习资料，这些学习资料不仅深入浅出，而且非常实用，让大家系统而高效地掌握AI大模型的各个知识点。

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】

AI大模型系统学习路线

在面对AI大模型开发领域的复杂与深入，精准学习显得尤为重要。一份系统的技术路线图，不仅能够帮助开发者清晰地了解从入门到精通所需掌握的知识点，还能提供一条高效、有序的学习路径。

但知道是一回事，做又是另一回事，初学者最常遇到的问题主要是理论知识缺乏、资源和工具的限制、模型理解和调试的复杂性，在这基础上，找到高质量的学习资源，不浪费时间、不走弯路，又是重中之重。

AI大模型入门到实战的视频教程+项目包

看视频学习是一种高效、直观、灵活且富有吸引力的学习方式，可以更直观地展示过程，能有效提升学习兴趣和理解力，是现在获取知识的重要途径

光学理论是没用的，要学会跟着一起敲，要动手实操，才能将自己的所学运用到实际当中去，这时候可以搞点实战案例来学习。

海量AI大模型必读的经典书籍（PDF）

阅读AI大模型经典书籍可以帮助读者提高技术水平，开拓视野，掌握核心技术，提高解决问题的能力，同时也可以借鉴他人的经验。对于想要深入学习AI大模型开发的读者来说，阅读经典书籍是非常有必要的。

600+AI大模型报告（实时更新）

这套包含640份报告的合集，涵盖了AI大模型的理论研究、技术实现、行业应用等多个方面。无论您是科研人员、工程师，还是对AI大模型感兴趣的爱好者，这套报告合集都将为您提供宝贵的信息和启示。

AI大模型面试真题+答案解析

我们学习AI大模型必然是想找到高薪的工作，下面这些面试题都是总结当前最新、最热、最高频的面试题，并且每道题都有详细的答案，面试前刷完这套面试题资料，小小offer，不在话下

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】