前言

最近大模型应用开发越来越火，但很多初学者面对RAG、Function Call、MCP这些概念时总是晕头转向。其实它们之间的关系并不复杂，用一个形象的比喻就能说清楚：

大模型是大脑，RAG是小抄，Function Call是左膀右臂，MCP是外包团队。

这篇文章会从概念讲起，然后带你一步步搭建一个完整的Demo。你可以一边读一边运行代码，真正理解它们是如何协同工作的。

---

第一部分：核心概念详解

1. 大模型调用 —— 应用的大脑

大模型是整个应用的核心决策者。无论是GPT-4、Claude还是DeepSeek，它们都充当了这个角色。

但大模型有几个天生的短板：

短板	说明
知识陈旧	训练数据截止到某个时间点，不知道最新发生的事情
无法访问私有数据	不知道你公司的内部文档、个人笔记
容易产生幻觉	遇到不懂的会编造答案

正是因为这些问题，我们才需要后面的三个模块来增强它。

2. RAG（检索增强生成）—— 给大模型配的"小抄"

RAG解决的是**“知识更新"和"私有数据接入”**的问题。

它的工作流程可以这样理解：

通俗来讲：就像考试允许你带小抄一样，RAG允许大模型在回答问题时"翻书"查资料。你先把公司文档、产品手册切碎存进向量数据库，用户提问时先去"翻书"找到相关内容，再连同问题一起交给大模型。这样大模型就能基于真实资料回答，大大减少胡编乱造。

3. Function Call —— 大模型的"左膀右臂"

RAG解决了"不知道"的问题，但大模型还有一个局限：它被困在服务器里，无法与外界交互。

比如用户问："北京明天天气怎么样？"大模型知道自己不知道，但它没法自己去查。这时候就需要Function Call。

通俗来说：Function Call就像大模型的一双手。当它需要获取外部信息或执行某个动作时，它会"举手"说：“我需要调用查天气这个工具”，应用层收到指令后去执行，再把结果还给它。

4. MCP（模型上下文协议）—— “外包团队”

如果说Function Call是你自己招人干活，那MCP就是把活外包给专业团队。

	Function Call	MCP
谁实现工具逻辑	开发者自己写代码	MCP Server提供方实现
接入成本	每个工具都要写一遍	即插即用，连接即可
维护责任	自己维护	由提供方维护
灵活性	高，完全可控	受限于Server提供的能力

MCP目前由Anthropic推动，越来越多的服务商开始提供MCP Server。未来大模型应用可能会像搭积木一样，只需要连接各种MCP Server就能快速构建复杂功能。

5. 四者协作全景图

---

第二部分：动手搭建Demo

理论讲完了，现在我们动手搭建一个完整的应用。这个Demo会实现：

1. ✅ 通过API调用大模型（DeepSeek）
2. ✅ RAG检索增强（本地产品手册）
3. ✅ Function Call工具调用（天气查询）
4. ✅ FastAPI提供HTTP接口

一、技术选型

组件	选择	说明
大模型API	DeepSeek	性价比高，国产，兼容OpenAI接口
向量数据库	ChromaDB	轻量级，嵌入式，适合入门
嵌入模型	all-MiniLM-L6-v2	轻量级，效果好，本地运行
Web框架	FastAPI	简单易用，自带API文档
开发语言	Python 3.10+	生态最丰富

二、项目结构

llm-app-demo/
├── .env                      # 环境变量
├── requirements.txt          # 依赖清单
├── main.py                   # FastAPI主服务
├── rag/
│   ├── __init__.py
│   ├── embedder.py           # 文本向量化
│   └── vector_store.py       # 向量数据库操作
├── tools/
│   ├── __init__.py
│   └── weather.py            # 天气工具（Function Call）
├── data/
│   └── product_manual.txt    # 知识库文档
└── prompts/
    └── system_prompt.py      # 系统提示词

三、环境准备

3.1 requirements.txt

fastapi==0.115.6
uvicorn==0.34.0
openai==1.59.3
chromadb==0.5.23
sentence-transformers==3.3.1
python-dotenv==1.0.1
pydantic==2.10.3

3.2 .env

DEEPSEEK_API_KEY=sk-你的DeepSeek密钥
DEEPSEEK_BASE_URL=https://api.deepseek.com/v1

💡 获取API Key：前往 DeepSeek官网 注册并创建API Key。

3.3 知识库文档 (data/product_manual.txt)

产品名称：智能助手Pro

核心功能：
1. 语音控制：支持中英文语音指令，识别准确率98%
2. 智能提醒：可设置日程提醒、吃药提醒、会议提醒
3. 健康监测：支持心率、血氧、睡眠质量监测
4. 智能家居联动：可连接米家、苹果HomeKit等平台

常见问题：
Q: 如何连接WiFi？
A: 在设备屏幕上依次点击"设置 → 网络 → WiFi"，选择您的网络并输入密码即可。

Q: 电池续航多久？
A: 正常使用情况下，续航可达7天；待机状态下可达15天。

Q: 支持哪些语言？
A: 目前支持中文、英文、日文、韩文四种语言。

售后服务：
- 质保期：12个月
- 客服电话：400-888-9999
- 工作时间：周一至周日 9:00-21:00

四、核心代码

4.1 向量嵌入模块 (rag/embedder.py)

from sentence_transformers import SentenceTransformer
import chromadb
from chromadb.config import Settings
from typing import List, Dict
import uuid


class Embedder:
    """文本向量化器"""
    def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
        self.model = SentenceTransformer(model_name)
    
    def embed(self, texts: List[str]) -> List[List[float]]:
        """将文本列表转换为向量"""
        return self.model.encode(texts, convert_to_numpy=True).tolist()


class VectorStore:
    """向量数据库管理"""
    def __init__(self, collection_name: str = "knowledge_base"):
        self.client = chromadb.PersistentClient(
            path="./chroma_db",
            settings=Settings(anonymized_telemetry=False)
        )
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            metadata={"hnsw:space": "cosine"}
        )
        self.embedder = Embedder()
    
    def add_documents(self, documents: List[str], metadatas: List[Dict] = None):
        """添加文档到向量库"""
        if metadatas is None:
            metadatas = [{}] * len(documents)
        
        ids = [str(uuid.uuid4()) for _ in documents]
        embeddings = self.embedder.embed(documents)
        
        self.collection.add(
            ids=ids,
            documents=documents,
            embeddings=embeddings,
            metadatas=metadatas
        )
        return ids
    
    def search(self, query: str, top_k: int = 3) -> List[Dict]:
        """检索最相关的文档片段"""
        query_embedding = self.embedder.embed([query])[0]
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=top_k
        )
        
        documents = []
        if results['documents'] and results['documents'][0]:
            for i, doc in enumerate(results['documents'][0]):
                documents.append({
                    'content': doc,
                    'metadata': results['metadatas'][0][i] if results['metadatas'] else {}
                })
        return documents

4.2 工具模块 (tools/weather.py)

import json
from typing import Dict, Any

# 天气工具定义（告诉大模型这个工具怎么用）
WEATHER_TOOL = {
    "type": "function",
    "function": {
        "name": "get_current_weather",
        "description": "获取指定城市的当前天气信息",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "城市名称，如：北京、上海、深圳"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "温度单位，摄氏或华氏"
                }
            },
            "required": ["city"]
        }
    }
}

# 工具列表（可注册多个）
TOOLS = [WEATHER_TOOL]


def execute_weather(city: str, unit: str = "celsius") -> Dict[str, Any]:
    """执行天气查询（模拟数据，生产环境替换为真实API）"""
    mock_data = {
        "北京": {"temperature": 28, "condition": "晴", "humidity": 45},
        "上海": {"temperature": 32, "condition": "多云", "humidity": 60},
        "深圳": {"temperature": 30, "condition": "阵雨", "humidity": 75},
        "广州": {"temperature": 33, "condition": "晴转多云", "humidity": 55},
        "杭州": {"temperature": 29, "condition": "小雨", "humidity": 70},
    }
    
    data = mock_data.get(city, {"temperature": 25, "condition": "未知", "humidity": 50})
    unit_symbol = "°C" if unit == "celsius" else "°F"
    temp = data["temperature"] if unit == "celsius" else data["temperature"] * 9/5 + 32
    
    return {
        "city": city,
        "temperature": round(temp, 1),
        "unit": unit_symbol,
        "condition": data["condition"],
        "humidity": data["humidity"]
    }


def handle_tool_call(tool_name: str, arguments: Dict) -> Dict:
    """工具调用分发器"""
    if tool_name == "get_current_weather":
        return execute_weather(**arguments)
    return {"error": f"未知工具: {tool_name}"}

4.3 系统提示词 (prompts/system_prompt.py)

SYSTEM_PROMPT = """
你是一个智能客服助手，名字叫"小智"。

## 职责
1. 回答用户关于产品的问题，基于提供的知识库内容
2. 帮助用户查询天气（通过调用工具）

## 规则
- 产品相关问题：优先使用知识库信息回答
- 知识库无相关信息：诚实告知"暂时没有找到"
- 天气等实时信息：使用get_current_weather工具
- 回答要友好、专业、简洁
"""

4.4 RAG初始化 (rag/init.py)

from .vector_store import VectorStore


def init_knowledge_base(file_path: str = "data/product_manual.txt"):
    """初始化知识库"""
    with open(file_path, 'r', encoding='utf-8') as f:
        content = f.read()
    
    # 按空行切分成段落
    chunks = [chunk.strip() for chunk in content.split('\n\n') if chunk.strip()]
    
    store = VectorStore()
    store.add_documents(chunks)
    print(f"✅ 知识库初始化完成，共导入 {len(chunks)} 个文档片段")
    return store


vector_store = None

def get_vector_store():
    global vector_store
    if vector_store is None:
        vector_store = VectorStore()
    return vector_store

4.5 主服务 (main.py)

import os
import json
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from openai import OpenAI
from dotenv import load_dotenv
from typing import List, Dict, Any

from rag import get_vector_store, init_knowledge_base
from tools.weather import TOOLS, handle_tool_call
from prompts.system_prompt import SYSTEM_PROMPT

load_dotenv()

# ====== 初始化 ======
app = FastAPI(title="大模型应用Demo")

client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL")
)

# 首次运行初始化知识库
try:
    init_knowledge_base()
except Exception as e:
    print(f"⚠️ 知识库初始化跳过: {e}")

vector_store = get_vector_store()


# ====== 数据模型 ======
class ChatRequest(BaseModel):
    message: str
    user_id: str = "default"


class ChatResponse(BaseModel):
    answer: str
    sources: List[str] = []
    tool_calls: List[Dict] = []


# ====== 核心逻辑 ======
def process_chat(message: str) -> Dict[str, Any]:
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    
    # 1. RAG检索判断
    rag_keywords = ["产品", "功能", "怎么", "如何", "售后", "质保", "连接", "支持"]
    need_rag = any(kw in message for kw in rag_keywords)
    
    sources = []
    if need_rag:
        results = vector_store.search(message, top_k=2)
        if results:
            context = "\n\n".join([r['content'] for r in results])
            sources = [r['content'][:80] + "..." for r in results]
            messages.append({
                "role": "user",
                "content": f"【知识库参考】\n{context}\n\n【问题】\n{message}"
            })
        else:
            messages.append({"role": "user", "content": message})
    else:
        messages.append({"role": "user", "content": message})
    
    # 2. 调用大模型
    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=messages,
        tools=TOOLS if not need_rag else [],
        tool_choice="auto" if not need_rag else "none",
        temperature=0.7,
    )
    
    assistant = response.choices[0].message
    tool_calls = []
    
    # 3. 处理Function Call
    if assistant.tool_calls:
        messages.append(assistant)
        for tc in assistant.tool_calls:
            name = tc.function.name
            args = json.loads(tc.function.arguments)
            result = handle_tool_call(name, args)
            tool_calls.append({"tool": name, "arguments": args, "result": result})
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": json.dumps(result, ensure_ascii=False)
            })
        
        final = client.chat.completions.create(
            model="deepseek-chat",
            messages=messages,
            temperature=0.7,
        )
        answer = final.choices[0].message.content
    else:
        answer = assistant.content
    
    return {"answer": answer, "sources": sources, "tool_calls": tool_calls}


# ====== API ======
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    try:
        result = process_chat(request.message)
        return ChatResponse(**result)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/health")
async def health():
    return {"status": "ok"}


@app.get("/rag/search")
async def search(query: str):
    return {"query": query, "results": vector_store.search(query, top_k=3)}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

五、启动与测试

5.1 启动服务

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

# 启动
python main.py

看到以下输出表示成功：

✅ 知识库初始化完成，共导入 8 个文档片段
INFO:     Uvicorn running on http://0.0.0.0:8000

5.2 测试RAG（产品问题）

curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "智能助手Pro怎么连接WiFi？"}'

返回结果：

{
  "answer": "在设备屏幕上依次点击\"设置 → 网络 → WiFi\"，选择您的网络并输入密码即可连接WiFi。",
  "sources": ["在设备屏幕上依次点击...", "产品名称：智能助手Pro..."],
  "tool_calls": []
}

5.3 测试Function Call（天气查询）

curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "北京今天天气怎么样？"}'

返回结果：

{
  "answer": "北京今天天气晴朗，气温28°C，湿度45%，非常适合户外活动。",
  "sources": [],
  "tool_calls": [
    {
      "tool": "get_current_weather",
      "arguments": {"city": "北京", "unit": "celsius"},
      "result": {"city": "北京", "temperature": 28, "unit": "°C", "condition": "晴", "humidity": 45}
    }
  ]
}

5.4 混合场景测试

curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "我想买智能助手Pro，上海今天天气如何？"}'

这时大模型会同时结合知识库信息和天气工具，给出综合回答。

---

六、完整请求处理时序图

---

七、MCP集成展望

当前的Demo使用的是Function Call模式。如果我们要升级到MCP模式，代码改动大致如下：

# 未来的MCP版本（示意）
from mcp import MCPClient

mcp_client = MCPClient()
# 连接一个天气MCP Server（由第三方提供）
weather_server = mcp_client.connect("https://weather.mcp.server")

# 大模型可以直接使用Server提供的工具
# 无需自己实现execute_weather函数

MCP带来的好处是显而易见的：

---

八、运行效果总结

用户问题类型	触发模块	示例	回答来源
产品知识类	RAG	“怎么连接WiFi？”	产品手册
实时信息类	Function Call	“今天天气如何？”	外部API
混合类	RAG + Function Call	“产品续航多久？顺便查下天气”	知识库 + API
闲聊类	大模型	“你好”	模型自身

---

九、写在最后

通过这篇文章，我们完成了：

1. 概念理解：用"大脑、小抄、左膀右臂、外包团队"的比喻理清了RAG、Function Call、MCP的关系
2. 完整Demo：从零搭建了一个包含RAG和Function Call的应用
3. 代码实践：所有代码可运行，方便你边学边练

接下来你可以继续探索的方向：

🔹 接入更多工具（发送邮件、查询数据库、调用其他API）
🔹 替换嵌入模型（BGE、OpenAI embedding等）
🔹 增加对话记忆（多轮对话上下文管理）
🔹 集成MCP客户端（关注Anthropic的MCP生态）
🔹 添加前端界面（Streamlit / Gradio快速构建）

大模型应用开发的门槛正在快速降低，希望这篇文章能帮你迈出第一步。

如果觉得有用，欢迎点赞、收藏、评论！ 🚀