本文详细介绍了大模型开发中的核心技术栈，包括MCP协议（连接AI模型与外部数据源）、LangChain框架（构建LLM应用）和LangGraph（多智能体工作流）。文章阐述了这些技术之间的关系、项目结构、配置文件和两种MCP集成方式（直接使用MCP Python客户端和使用langchain-mcp-adapters），为开发者提供了构建强大AI应用系统的技术路径和最佳实践。

核心概念解析

1. MCP (Model Context Protocol)

定义：模型上下文协议，是一个开放标准，用于连接AI模型与外部数据源和工具。

核心特性：

  • 标准化接口：统一的协议规范，支持多种数据源
  • 工具抽象：将外部功能抽象为工具，供AI模型调用
  • 资源管理：管理文件、数据库连接等资源
  • 流式支持：支持流式数据处理

协议组成：

{
  "mcpServers": {
    "server-name": {
      "command": "启动命令",
      "env": {
        "TRANSPORT": "stdio",
        "DSN": "数据源连接串",
        "READONLY": "true"
      }
    }
  }
}

2. LangChain

定义：一个用于构建LLM应用的框架，提供组件化、可组合的AI应用开发工具。

核心组件：

  • Chains：链式调用多个组件
  • Agents：智能体，能够使用工具
  • Tools：工具，执行特定功能
  • Memory：记忆组件，存储对话历史
  • Callbacks：回调机制，监控执行过程

3. LangGraph

定义：基于LangChain的状态图框架，用于构建复杂的多智能体工作流。

核心特性：

  • 状态管理：管理智能体间的状态传递
  • 条件路由：根据条件决定下一步执行
  • 并行处理：支持并行执行多个智能体
  • 可视化：提供工作流图的可视化

4. MCP Server

定义：实现MCP协议的服务器，提供特定数据源或功能的访问接口。

常见MCP Server：

  • @bytebase/dbhub：数据库连接器（MySQL、PostgreSQL等）
  • mcp-server-filesystem：文件系统访问
  • mcp-server-http：HTTP API访问
  • mcp-server-weather：天气数据服务

5. MCP Client

定义：MCP协议的客户端实现，用于连接和调用MCP Server。

主要实现：

  • mcp-python：Python官方客户端
  • mcp-cli：命令行客户端
  • langchain-mcp-adapters：LangChain集成客户端

6. langchain-mcp-adapters

定义：LangChain官方提供的MCP适配器，将MCP工具转换为LangChain工具。

核心功能：

  • 工具转换：MCP工具 → LangChain工具
  • 服务器管理：管理多个MCP服务器连接
  • 资源处理：处理MCP资源（文件、数据等）
  • 异步支持：支持异步操作

技术架构关系

整体架构图

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   LangGraph     │    │   LangChain     │    │   MCP Client    │
│   (工作流编排)    │◄──►│   (智能体框架)    │◄──►│   (协议客户端)    │
└─────────────────┘    └─────────────────┘    └─────────────────┘
                                │                        │
                                ▼                        ▼
                       ┌─────────────────┐    ┌─────────────────┐
                       │ MCP Adapters    │    │   MCP Protocol  │
                       │ (工具转换层)      │◄──►│   (协议层)       │
                       └─────────────────┘    └─────────────────┘
                                                        │
                                                        ▼
                                               ┌─────────────────┐
                                               │   MCP Servers   │
                                               │ (数据源服务)      │
                                               └─────────────────┘

数据流向

  1. LangGraph 编排工作流，调用 LangChain 智能体
  2. LangChain 智能体通过 MCP Adapters 获取工具
  3. MCP Adapters 将MCP工具转换为LangChain工具
  4. MCP Client 通过协议与 MCP Server 通信
  5. MCP Server 访问实际的数据源（数据库、API、文件等）

项目结构概览

核心目录结构

intelligent_modeling2/
├── agents/                    # 智能体定义
│   ├── base_agent.py         # 基础智能体类
│   ├── requirement_analyst.py # 需求分析智能体
│   ├── data_modeler.py       # 数据建模智能体
│   └── ...
├── utils/                     # 工具和配置
│   ├── env_config.py         # 环境变量配置管理
│   ├── mcp_config_manager.py # MCP配置管理器
│   ├── cache_manager.py      # 缓存管理器
│   ├── logger.py             # 统一日志系统
│   ├── graph.py              # LangGraph工作流定义
│   ├── langchain_mcp_integration.py # LangChain MCP集成
│   ├── mcp_database_connector.py    # MCP数据库连接器
│   └── ...
├── examples/                  # 示例和演示
│   ├── data_warehouse_modeling.py   # 数仓建模主程序
│   ├── simple_data_modeling.py      # 简化建模流程
│   ├── mcp_bytebase_integration.py  # MCP集成示例
│   ├── research_assistant.py        # 研究助手
│   └── ...
├── tests/                     # 测试文件
│   ├── test_mcp_integration.py      # MCP集成测试
│   └── ...
├── docs/                      # 文档
│   ├── MCP_Integration_Guide.md     # MCP集成指南
│   ├── Simple_Modeling_Guide.md     # 简化建模指南
│   └── ...
├── cache/                     # 缓存目录
├── outputs/                   # 输出目录
├── mcp_config.json           # MCP服务器配置
├── env.example               # 环境变量示例
└── main.py                   # 主程序入口

配置文件说明

1. mcp_config.json - MCP服务器配置
{
    "mcpServers": {
        "dbhub-mysql-demo": {
            "command": "npx @bytebase/dbhub",
            "env": {
                "TRANSPORT": "stdio",
                "DSN": "mysql://user:pass@host:port/db",
                "READONLY": "true"
            }
        }
    }
}
2. env.example - 环境变量配置
# LLM模型配置
LLM_TEMPERATURE=0.1
LLM_MAX_TOKENS=4000
LLM_TIMEOUT=300
LLM_RETRY_COUNT=5
LLM_RETRY_DELAY=5

# 工作流配置
WORKFLOW_RECURSION_LIMIT=50
WORKFLOW_MAX_LOOPS_PER_STAGE=3

# MCP配置
MCP_INIT_TIMEOUT=5.0
MCP_READ_TIMEOUT=10.0
MCP_MYSQL_CONNECT_TIMEOUT=30
MCP_MYSQL_READ_TIMEOUT=30
MCP_MYSQL_WRITE_TIMEOUT=30

# API密钥
OPENAI_API_KEY=your_openai_api_key_here
SILICONFLOW_API_KEY=your_siliconflow_api_key_here
SILICONFLOW_API_BASE=https://api.siliconflow.cn/v1

MCP集成方式

方式一：直接使用MCP Python客户端

优点
  • 简单直接：无需额外框架
  • 性能好：直接协议通信
  • 灵活性高：完全控制通信过程
缺点
  • 需要手动管理：工具注册、错误处理等
  • 集成复杂：需要自己处理与智能体的集成
实现示例
from mcp import MCPClient
import asyncio

class DirectMCPClient:
    def __init__(self):
        self.clients = {}

    async def connect_database(self, name: str, dsn: str):
        """连接数据库"""
        client = MCPClient(
            command="npx @bytebase/dbhub",
            env={
                "TRANSPORT": "stdio",
                "DSN": dsn,
                "READONLY": "true"
            }
        )
        self.clients[name] = client
        return client

    async def get_schema(self, db_name: str, table_name: str = None):
        """获取schema信息"""
        client = self.clients.get(db_name)
        if not client:
            return None

        if table_name:
            return await client.get_table_schema(table_name)
        else:
            return await client.list_tables()

    async def execute_query(self, db_name: str, query: str):
        """执行查询"""
        client = self.clients.get(db_name)
        if not client:
            return None

        return await client.execute_sql(query)

# 使用示例
async def example_usage():
    client = DirectMCPClient()

    # 连接MySQL
    await client.connect_database(
        "mysql-demo",
        "mysql://user:pass@host:port/db"
    )

    # 获取schema
    schema = await client.get_schema("mysql-demo")
    print(f"Tables: {schema}")

    # 执行查询
    result = await client.execute_query(
        "mysql-demo", 
        "SELECT COUNT(*) FROM orders"
    )
    print(f"Result: {result}")

方式二：使用langchain-mcp-adapters

优点
  • 无缝集成：与LangChain生态系统完美集成
  • 工具自动转换：MCP工具自动转换为LangChain工具
  • 智能体友好：智能体可以直接使用工具
  • 标准化：遵循LangChain工具标准
缺点
  • 依赖较重：需要安装LangChain相关包
  • 学习成本：需要了解LangChain概念
实现示例
from langchain_mcp import MCPToolkit
from langchain_mcp.server import MCPServer
from langchain_core.tools import BaseTool
import asyncio

class LangChainMCPManager:
    def __init__(self):
        self.servers = {}
        self.toolkits = {}
        self.tools = []

    async def register_database(self, name: str, dsn: str):
        """注册数据库MCP服务器"""
        server = MCPServer(
            name=name,
            command="npx @bytebase/dbhub",
            env={
                "TRANSPORT": "stdio",
                "DSN": dsn,
                "READONLY": "true"
            }
        )

        self.servers[name] = server
        toolkit = MCPToolkit(server)
        self.toolkits[name] = toolkit

        # 获取工具
        tools = await toolkit.get_tools()
        self.tools.extend(tools)

        return tools

    def get_tools_for_agent(self):
        """获取供智能体使用的工具"""
        return self.tools

    async def execute_tool(self, tool_name: str, **kwargs):
        """执行工具"""
        for tool in self.tools:
            if tool.name == tool_name:
                return await tool.ainvoke(kwargs)
        return None

# 在LangGraph中使用
from langchain_core.tools import tool
from langgraph import StateGraph, END

# 创建MCP管理器
mcp_manager = LangChainMCPManager()

# 注册数据库
await mcp_manager.register_database(
    "mysql-demo",
    "mysql://user:pass@host:port/db"
)

# 获取工具
tools = mcp_manager.get_tools_for_agent()

# 在智能体中使用
@tool
async def query_database(query: str):
    """查询数据库"""
    return await mcp_manager.execute_tool("query", sql=query)

# 创建工作流
workflow = StateGraph(AgentState)
workflow.add_node("query_agent", query_database)
workflow.add_edge("query_agent", END)

相关资源

官方文档

  • MCP协议规范
  • LangChain文档
  • LangGraph文档
  • langchain-mcp-adapters

常用MCP Server

  • @bytebase/dbhub - 数据库连接器
  • mcp-server-filesystem - 文件系统
  • mcp-server-http - HTTP API

示例项目

  • MCP Python客户端示例
  • LangChain MCP集成示例

总结

MCP协议为AI应用提供了强大的数据源集成能力，通过三种不同的集成方式，可以满足不同场景的需求：

  1. 直接MCP客户端：适合简单、直接的场景
  2. langchain-mcp-adapters：适合LangChain生态系统
  3. 自定义适配器：适合需要深度定制的场景

选择合适的集成方式，结合最佳实践，可以构建出强大、可靠的AI应用系统。

如何系统学习掌握AI大模型？

AI大模型作为人工智能领域的重要技术突破，正成为推动各行各业创新和转型的关键力量。抓住AI大模型的风口，掌握AI大模型的知识和技能将变得越来越重要。

学习AI大模型是一个系统的过程，需要从基础开始，逐步深入到更高级的技术。

这里给大家精心整理了一份全面的AI大模型学习资源，包括：AI大模型全套学习路线图（从入门到实战）、精品AI大模型学习书籍手册、视频教程、实战学习、面试题等，资料免费分享！

1. 成长路线图&学习规划

要学习一门新的技术，作为新手一定要先学习成长路线图，方向不对，努力白费。

这里，我们为新手和想要进一步提升的专业人士准备了一份详细的学习成长路线图和规划。可以说是最科学最系统的学习成长路线。

在这里插入图片描述

2. 大模型经典PDF书籍

书籍和学习文档资料是学习大模型过程中必不可少的，我们精选了一系列深入探讨大模型技术的书籍和学习文档，它们由领域内的顶尖专家撰写，内容全面、深入、详尽，为你学习大模型提供坚实的理论基础。（书籍含电子版PDF）

在这里插入图片描述

3. 大模型视频教程

对于很多自学或者没有基础的同学来说，书籍这些纯文字类的学习教材会觉得比较晦涩难以理解，因此，我们提供了丰富的大模型视频教程，以动态、形象的方式展示技术概念，帮助你更快、更轻松地掌握核心知识。

在这里插入图片描述

4. 大模型行业报告

行业分析主要包括对不同行业的现状、趋势、问题、机会等进行系统地调研和评估，以了解哪些行业更适合引入大模型的技术和应用，以及在哪些方面可以发挥大模型的优势。

在这里插入图片描述

5. 大模型项目实战

学以致用 ，当你的理论知识积累到一定程度，就需要通过项目实战，在实际操作中检验和巩固你所学到的知识，同时为你找工作和职业发展打下坚实的基础。

在这里插入图片描述

6. 大模型面试题

面试不仅是技术的较量，更需要充分的准备。

在你已经掌握了大模型技术之后，就需要开始准备面试，我们将提供精心整理的大模型面试题库，涵盖当前面试中可能遇到的各种技术问题，让你在面试中游刃有余。

在这里插入图片描述

全套的AI大模型学习资源已经整理打包，有需要的小伙伴可以微信扫描下方CSDN官方认证二维码，免费领取【保证100%免费】

# 人工智能
Logo
2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

cover

《The Landscape of Agentic Reinforcement Learning for LLMs: A Survey》

avatar 2048 AI社区
cover

C++11 智能指针的使⽤及其原理

avatar 2048 AI社区

Linux常用命令

本文介绍了Linux常用命令使用技巧，包括： 目录操作：ls/cd/mkdir/rm等命令详解 文件操作：cat/more/head/tail等查看命令 文件管理：cp/mv等复制移动命令 压缩解压：tar打包压缩命令 文本编辑：vim三种模式及基本操作 查找命令：find/grep文件及内容查找 各命令均配有常用选项说明和典型用法示例，特别强调了危险操作(如rm -rf)的注意事项。

avatar 2048 AI社区