深度实战：基于 LangGraph + MCP 的智能音乐推荐 Agent 全解析（从搭建到落地）

在 AI 与推荐系统结合的浪潮中，传统音乐推荐往往局限于“基于协同过滤”或“关键词匹配”，无法理解用户自然语言中的模糊需求（如“心情低落想听治愈系民谣”“运动时需要燃脂电子乐”）。而基于 LangGraph + MCP 构建的智能音乐推荐 Agent，恰好解决了这一痛点——它能听懂自然语言意图，通过标准化工具调用实现个性化推荐，还能输出可解释的推荐理由。

一、项目核心价值与技术选型深度解析

1. 项目定位与核心优势

• 自然语言意图理解：支持多维度需求（心情、场景、流派、相似歌曲、艺术家），无需用户手动筛选；
• 可解释推荐：每首歌附带推荐理由（如“节奏明快适合运动”“旋律治愈匹配低落心情”），提升用户信任度；
• 轻量化可扩展：基于 Streamlit 快速搭建 UI，支持本地数据与外部 API 切换，MCP 协议适配多工具；
• 技术栈灵活：LangGraph 编排工作流，支持替换 LLM（DeepSeek、Qwen、GPT 等），适配不同预算与场景。

2. 技术选型深层原因

技术组件	核心作用	选型深层原因
LangGraph	工作流编排与状态管理	支持复杂分支逻辑（多意图路由）、状态持久化，适合 Agent 任务拆解
LangChain	LLM 工具链与提示词管理	封装 LLM 调用、工具适配，降低与 MCP/外部 API 的集成成本
硅基流动（SiliconFlow）	LLM 服务提供	兼容 OpenAI API 格式，支持多模型切换，无需修改业务代码
Streamlit	轻量 Web UI 开发	快速迭代原型，支持组件化布局，无需前端框架（Vue/React）经验
MCP	工具调用标准化	统一对接音乐平台 API、本地工具，降低跨平台适配成本
Pydantic	数据校验与模型定义	确保 LLM 输出格式规范，避免推荐结果解析失败

3. 适用场景扩展

• 个人使用：本地部署打造专属音乐助手，支持自定义音乐库；
• 创业原型：快速验证 AI 推荐产品可行性，对接 Spotify/网易云 API 即可商业化；
• 企业内部工具：适配办公场景（如“会议室背景音乐推荐”“员工健身音乐合集”）；
• 教学/研究：学习 LangGraph 工作流编排、MCP 协议应用、意图识别技术。

二、快速上手：5 分钟跑通 + 常见问题速解

1. 环境准备

• 操作系统：Windows 10+/macOS 12+/Linux（Ubuntu 20.04+）；
• Python 版本：3.9-3.11（推荐 3.10，避免版本兼容问题）；
• 核心依赖版本锁定（避免依赖冲突）：

  langgraph==0.1.14
  langchain==0.1.10
  streamlit==1.32.2
  openai==1.13.3
  pydantic==2.6.1
  aiohttp==3.9.3
  python-dotenv==1.0.1

2. 完整部署步骤

步骤 1：克隆项目与安装依赖

# 克隆仓库（替换为实际仓库地址）
git clone https://github.com/imagist13/Muisc-Research.git
cd Muisc-Research

# 创建虚拟环境（推荐，避免依赖污染）
python -m venv venv
# 激活虚拟环境（Windows）
venv\Scripts\activate
# 激活虚拟环境（macOS/Linux）
source venv/bin/activate

# 安装依赖（指定版本避免冲突）
pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

步骤 2：配置文件详解（setting.json 全字段说明）

{
  "SILICONFLOW_API_KEY": "你的硅基流动 API Key", // 从硅基流动控制台获取
  "SILICONFLOW_BASE_URL": "https://api.siliconflow.cn/v1", // 固定地址
  "SILICONFLOW_CHAT_MODEL": "Qwen/Qwen2.5-72B-Instruct", // 可选：DeepSeek/DeepSeek-MoE-16B-Instruct
  "SILICONFLOW_TEMPERATURE": 0.3, // 推荐 0.2-0.5，值越低推荐越稳定
  "TAILYAPI_API_KEY": "你的 Tavily 搜索 API Key", // 用于补充实时音乐数据（可选）
  "TAILYAPI_BASE_URL": "https://api.tavily.com",
  "APP_NAME": "智能音乐推荐助手", // 自定义应用名称
  "SPOTIFY_CLIENT_ID": "你的 Spotify Client ID", // 对接 Spotify 时必填
  "SPOTIFY_CLIENT_SECRET": "你的 Spotify Client Secret",
  "LOCAL_MUSIC_DB_PATH": "data/music_database.json", // 本地音乐数据库路径
  "CACHE_EXPIRE_SECONDS": 3600, // 推荐结果缓存时间（1小时）
  "MAX_RECOMMEND_NUM": 10, // 单次最大推荐歌曲数
  "INTENT_RECOGNITION_PROMPT_PATH": "prompts/intent_recognition.txt" // 意图识别提示词路径
}

步骤 3：启动应用与验证

# 启动 Streamlit 应用
python run_music_app.py
# 或直接启动（跳过检查脚本）
streamlit run music_app.py --server.port 8501

• 访问地址：http://localhost:8501
• 验证场景：
  1. 输入“心情很好，推荐开心的流行乐”，查看推荐结果与理由；
  2. 点击侧边栏“运动”快捷按钮，验证场景化推荐；
  3. 输入“类似《晴天》的歌曲”，验证相似歌曲推荐。

3. 快速上手常见问题速解

问题现象	解决方案
启动报错“缺少 SILICONFLOW_API_KEY”	1. 硅基流动官网注册获取 API Key；2. 填入 setting.json
Streamlit 启动后无法访问	检查端口是否被占用，更换端口：--server.port 8502
推荐结果为空	1. 检查本地音乐数据库（data/music_database.json）是否有数据；2. 确保 LLM 模型配置正确
意图识别错误（如“运动”识别为“聊天”）	优化 prompts/intent_recognition.txt 中的提示词，增加意图示例

三、系统架构深度解析：模块交互与数据流转

1. 整体架构图（补充交互关系）

用户 → Streamlit UI（输入/展示） → MusicAgent 主入口 → LangGraph 工作流
                                      ↓
                          ┌─────────┬─────────┬─────────┬─────────┐
                          │  LLM 层  │ 工具层  │ 配置层  │ 数据层  │
                          │（意图识别）│（MCP/API）│（参数管理）│（本地/外部）│
                          └─────────┴─────────┴─────────┴─────────┘
                                      ↓
                          MCP Server → 外部工具（Spotify API/音乐搜索）

2. 核心模块职责与交互细节

（1）配置层（config/）：统一参数管理

• 核心文件：settings_loader.py
• 核心逻辑：
  1. 优先级：setting.json > 环境变量 > 默认值；
  2. 支持动态加载：修改配置后无需重启应用，下次请求自动生效；
  3. 关键功能：参数校验（如 API Key 格式、模型名称合法性）。

• 代码解析：

  def load_settings():
    # 加载 JSON 配置
    with open("setting.json", "r", encoding="utf-8") as f:
        json_config = json.load(f)
    # 合并环境变量（覆盖 JSON 配置）
    env_config = {
        key: os.getenv(key) for key in json_config if os.getenv(key) is not None
    }
    final_config = {**json_config, **env_config}
    # 参数校验
    validate_settings(final_config)
    return final_config

（2）LLM 层（llms/）：模型抽象与适配

• 核心文件：siliconflow_llm.py
• 设计思路：
  1. 兼容 OpenAI API 格式，替换模型时无需修改业务代码；
  2. 封装重试逻辑（API 超时、限流时自动重试）；
  3. 支持模型调优参数（temperature、max_tokens、top_p）。

• 关键代码：

  class SiliconFlowLLM:
    def __init__(self, api_key, base_url, model_name, temperature=0.3):
        self.client = OpenAI(
            api_key=api_key,
            base_url=base_url
        )
        self.model_name = model_name
        self.temperature = temperature
  
    def chat_completion(self, messages):
        # 重试逻辑（最多 3 次）
        for _ in range(3):
            try:
                response = self.client.chat.completions.create(
                    model=self.model_name,
                    messages=messages,
                    temperature=self.temperature,
                    response_format={"type": "json_object"}  # 强制 JSON 输出
                )
                return json.loads(response.choices[0].message.content)
            except Exception as e:
                log.error(f"LLM 调用失败：{e}")
                time.sleep(1)
        raise Exception("LLM 调用多次失败，请检查 API Key 或网络")

（3）工作流层（graphs/）：LangGraph 状态与路由

• 核心概念：
  • 状态（State）：存储用户输入、意图类型、推荐结果、错误信息；
  • 节点（Node）：封装具体逻辑（意图识别、搜索、推荐、结果生成）；
  • 路由（Route）：根据意图类型分发到对应节点。

• 状态定义代码：

  from langgraph.graph import StateGraph, MessagesState
  from pydantic import BaseModel
  
  # 自定义状态（扩展 MessagesState）
  class MusicAgentState(MessagesState):
    intent: str = None  # 意图类型（recommend_by_mood/activity/genre 等）
    recommendations: list = []  # 推荐结果列表
    error: str = None  # 错误信息

• 核心节点实现：

  def analyze_intent(state: MusicAgentState) -> MusicAgentState:
    """意图识别节点：从用户输入中提取意图"""
    user_input = state.messages[-1].content
    # 调用 LLM 识别意图
    llm = SiliconFlowLLM(**settings)
    prompt = load_prompt("prompts/intent_recognition.txt")
    messages = [
        {"role": "system", "content": prompt},
        {"role": "user", "content": user_input}
    ]
    result = llm.chat_completion(messages)
    state.intent = result["intent"]
    state.error = result.get("error")
    return state
  
  def route_by_intent(state: MusicAgentState) -> str:
    """路由节点：根据意图分发到对应处理节点"""
    if state.error:
        return "handle_error"
    match state.intent:
        case "recommend_by_mood":
            return "recommend_by_mood"
        case "recommend_by_activity":
            return "recommend_by_activity"
        case "search_music":
            return "search_music"
        case "general_chat":
            return "general_chat"
        case _:
            return "handle_unknown_intent"

• 工作流构建：

  def build_music_graph() -> StateGraph:
    graph = StateGraph(MusicAgentState)
    # 添加节点
    graph.add_node("analyze_intent", analyze_intent)
    graph.add_node("recommend_by_mood", recommend_by_mood)
    graph.add_node("recommend_by_activity", recommend_by_activity)
    graph.add_node("search_music", search_music)
    graph.add_node("generate_result", generate_result)
    graph.add_node("handle_error", handle_error)
    # 定义边（路由逻辑）
    graph.set_entry_point("analyze_intent")
    graph.add_conditional_edges("analyze_intent", route_by_intent)
    # 所有处理节点最终指向结果生成
    graph.add_edge("recommend_by_mood", "generate_result")
    graph.add_edge("recommend_by_activity", "generate_result")
    graph.add_edge("search_music", "generate_result")
    graph.add_edge("general_chat", "generate_result")
    graph.add_edge("handle_error", "generate_result")
    graph.add_edge("handle_unknown_intent", "generate_result")
    return graph.compile()

（4）工具层（tools/）：MCP 适配与外部调用

工具层是连接 Agent 与外部服务的核心：

• 工具分类：
  • 本地工具：本地音乐搜索、相似歌曲匹配（基于 TF-IDF 或余弦相似度）；
  • MCP 工具：对接外部音乐 API（Spotify、网易云）、音乐评分查询。

• MCP Server 开发（音乐搜索工具示例）：

  # mcp/music_search_server.py
  from mcp.server import Server, Tool
  from mcp.types import InputSchema, OutputSchema
  from pydantic import BaseModel
  
  # 输入/输出模型
  class MusicSearchInput(BaseModel):
    keyword: str  # 搜索关键词（歌曲名/艺术家）
    genre: str = None  # 流派筛选（可选）
  
  class MusicSearchOutput(BaseModel):
    songs: list[dict]  # 搜索结果：[{"title": "", "artist": "", "album": ""}]
  
  # 工具实现
  def search_music(input_data: MusicSearchInput) -> MusicSearchOutput:
    # 对接本地数据库或外部 API
    with open(settings["LOCAL_MUSIC_DB_PATH"], "r", encoding="utf-8") as f:
        music_db = json.load(f)
    # 筛选逻辑
    results = [
        song for song in music_db
        if input_data.keyword.lower() in song["title"].lower() 
        and (input_data.genre is None or song["genre"] == input_data.genre)
    ]
    return MusicSearchOutput(songs=results[:10])
  
  # 注册 MCP Server
  server = Server()
  search_tool = Tool(
    name="music_search",
    description="根据关键词和流派搜索音乐",
    input_schema=InputSchema(model=MusicSearchInput),
    output_schema=OutputSchema(model=MusicSearchOutput),
    handler=search_music
  )
  server.register_tool(search_tool)
  
  # 启动 MCP Server
  if __name__ == "__main__":
    import uvicorn
    uvicorn.run(server.fastapi_app, host="0.0.0.0", port=8000)

• Agent 调用 MCP 工具代码：

  def search_music(state: MusicAgentState) -> MusicAgentState:
    """搜索节点：调用 MCP 工具搜索音乐"""
    user_input = state.messages[-1].content
    # 提取搜索关键词（通过 LLM 解析）
    llm = SiliconFlowLLM(**settings)
    prompt = "提取用户输入中的音乐搜索关键词和流派（如无则为 None），返回 JSON：{\"keyword\": \"\", \"genre\": \"\"}"
    messages = [{"role": "user", "content": f"{prompt}\n用户输入：{user_input}"}]
    search_params = llm.chat_completion(messages)
    # 调用 MCP 工具
    mcp_client = Client(base_url="http://localhost:8000")
    try:
        response = mcp_client.call_tool("music_search", search_params)
        state.recommendations = response.content[0].text["songs"]
    except Exception as e:
        state.error = f"搜索失败：{str(e)}"
    return state

（5）前端层（music_app.py）：交互体验优化

• 核心组件：
  1. 侧边栏：快捷按钮（心情/场景/流派）、配置状态显示；
  2. 主区域：用户输入框、推荐结果卡片（含歌曲信息、推荐理由）、历史记录；
  3. 结果卡片：使用 Streamlit 的 card 组件，支持点击播放链接。

• 关键代码：

  def render_recommendation_cards(recommendations):
    """渲染推荐结果卡片"""
    for idx, song in enumerate(recommendations, 1):
        with st.card(f"推荐歌曲 #{idx}"):
            col1, col2 = st.columns([1, 2])
            with col1:
                st.image(song.get("cover_url", "default_cover.png"), use_column_width=True)
            with col2:
                st.subheader(song["title"])
                st.text(f"艺术家：{song['artist']}")
                st.text(f"流派：{song['genre']}")
                st.text(f"推荐理由：{song['reason']}")
                # 播放链接（对接 Spotify 时生效）
                if song.get("spotify_url"):
                    st.link_button("Spotify 播放", song["spotify_url"])

四、核心流程：从意图识别到推荐结果的全链路拆解

1. 核心时序图（用户输入“运动时需要燃脂电子乐”）

用户 → 输入需求 → Streamlit UI → 提交请求 → MusicAgent
                          ↓
MusicAgent → 调用 LangGraph 工作流 → 分析意图节点（analyze_intent）
                          ↓
LLM → 识别意图为“recommend_by_activity” → 路由节点（route_by_intent）
                          ↓
推荐节点（recommend_by_activity） → 调用 MCP 工具搜索“燃脂电子乐”
                          ↓
外部 API/本地数据库 → 返回符合条件的歌曲列表 → 结果生成节点（generate_result）
                          ↓
LLM → 为每首歌生成推荐理由 → Streamlit UI → 渲染结果卡片

2. 关键环节详解

（1）意图识别：提示词优化技巧

意图识别的准确性直接影响推荐效果，推荐使用“示例引导+格式约束”的提示词：

# prompts/intent_recognition.txt
你是音乐推荐 Agent 的意图识别助手，需从用户输入中提取意图类型，返回 JSON 格式（必须包含 intent 字段）。
支持的意图类型：
1. recommend_by_mood：用户提到心情（如开心、悲伤、治愈）；
2. recommend_by_activity：用户提到场景/活动（如运动、工作、睡觉）；
3. recommend_by_genre：用户提到音乐流派（如民谣、电子乐、摇滚）；
4. recommend_by_artist：用户提到艺术家/歌手（如周杰伦、Taylor Swift）；
5. recommend_by_similar：用户提到“类似/像XX歌曲”；
6. search_music：用户明确搜索歌曲/专辑；
7. general_chat：不涉及音乐推荐的闲聊。

示例：
用户输入：“心情低落想听治愈系民谣” → {"intent": "recommend_by_mood", "mood": "治愈"}
用户输入：“运动时需要燃脂的电子乐” → {"intent": "recommend_by_activity", "activity": "运动", "genre": "电子乐"}
用户输入：“类似《晴天》的歌曲” → {"intent": "recommend_by_similar", "song": "晴天"}

注意：
- 若无法识别意图，intent 设为 "unknown"；
- 额外提取相关参数（如 mood、activity、genre），无则留空；
- 严格返回 JSON，不添加额外描述。

（2）推荐逻辑：结合用户需求与音乐特征

推荐节点的核心是“筛选符合条件的歌曲 + 个性化排序”，示例代码：

def recommend_by_activity(state: MusicAgentState) -> MusicAgentState:
    """根据活动场景推荐音乐"""
    # 提取活动类型和额外参数
    user_input = state.messages[-1].content
    llm = SiliconFlowLLM(**settings)
    prompt = "提取用户输入中的活动类型和音乐偏好（如流派、节奏），返回 JSON：{\"activity\": \"\", \"genre\": \"\", \"tempo\": \"\"}"
    params = llm.chat_completion([{"role": "user", "content": f"{prompt}\n{user_input}"}])
    activity = params["activity"]
    genre = params.get("genre")
    tempo = params.get("tempo", "fast")  # 默认快节奏（运动场景）

    # 从本地数据库/外部 API 筛选歌曲
    with open(settings["LOCAL_MUSIC_DB_PATH"], "r", encoding="utf-8") as f:
        music_db = json.load(f)
    # 筛选逻辑：活动匹配 + 流派匹配 + 节奏匹配
    filtered_songs = [
        song for song in music_db
        if song["activity"] == activity 
        and (genre is None or song["genre"] == genre)
        and song["tempo"] == tempo
    ]
    # 排序：按流行度降序
    filtered_songs.sort(key=lambda x: x["popularity"], reverse=True)
    state.recommendations = filtered_songs[:settings["MAX_RECOMMEND_NUM"]]
    return state

（3）结果生成：可解释性优化

推荐结果需包含“歌曲信息 + 推荐理由”，让用户理解“为什么推荐这首歌”：

def generate_result(state: MusicAgentState) -> MusicAgentState:
    """生成可解释的推荐结果"""
    if state.error:
        result = f"推荐失败：{state.error}"
    elif not state.recommendations:
        result = "未找到符合条件的歌曲，可尝试调整需求描述～"
    else:
        # 生成推荐理由
        llm = SiliconFlowLLM(**settings)
        prompt = load_prompt("prompts/generate_result.txt")
        songs_str = json.dumps([{"title": s["title"], "artist": s["artist"], "genre": s["genre"]} for s in state.recommendations])
        messages = [
            {"role": "system", "content": prompt},
            {"role": "user", "content": f"用户需求：{state.messages[-1].content}\n推荐歌曲：{songs_str}"}
        ]
        result = llm.chat_completion(messages)
        # 合并歌曲信息与推荐理由
        for song, reason in zip(state.recommendations, result["reasons"]):
            song["reason"] = reason

    # 添加到状态，用于前端展示
    state.messages.append({"role": "assistant", "content": json.dumps(state.recommendations)})
    return state

五、进阶实战：MCP + Spotify API 对接全流程

1. Spotify 开发者账号准备

步骤 1：创建应用

1. 访问 Spotify for Developers，注册账号并创建应用；
2. 记录 Client ID 和 Client Secret（填入 setting.json）；
3. 在“Redirect URIs”中添加 http://localhost:8501/callback（用于授权）。

步骤 2：获取访问令牌（Authorization Code Flow）

Spotify API 需授权才能调用，核心流程：

1. 前端引导用户授权，获取 Authorization Code；
2. 后端用 Code 兑换 Access Token 和 Refresh Token；
3. 用 Access Token 调用 API，过期后用 Refresh Token 刷新。

2. MCP Server 对接 Spotify API

步骤 1：开发 Spotify 音乐搜索 MCP Server

# mcp/spotify_search_server.py
from mcp.server import Server, Tool
from mcp.types import InputSchema, OutputSchema
from pydantic import BaseModel
import spotipy
from spotipy.oauth2 import SpotifyOAuth

# 输入/输出模型
class SpotifySearchInput(BaseModel):
    keyword: str
    genre: str = None
    limit: int = 10

class SpotifySearchOutput(BaseModel):
    songs: list[dict] = []

# Spotify 客户端初始化
def init_spotify_client():
    return spotipy.Spotify(
        auth_manager=SpotifyOAuth(
            client_id=settings["SPOTIFY_CLIENT_ID"],
            client_secret=settings["SPOTIFY_CLIENT_SECRET"],
            redirect_uri="http://localhost:8501/callback",
            scope="user-read-private user-read-email playlist-modify-private"
        )
    )

# 工具实现
def spotify_search(input_data: SpotifySearchInput) -> SpotifySearchOutput:
    try:
        sp = init_spotify_client()
        # 搜索歌曲
        query = f"{input_data.keyword} {'genre:' + input_data.genre if input_data.genre else ''}"
        results = sp.search(q=query, type="track", limit=input_data.limit)
        # 解析结果
        songs = []
        for track in results["tracks"]["items"]:
            songs.append({
                "title": track["name"],
                "artist": ", ".join([a["name"] for a in track["artists"]]),
                "album": track["album"]["name"],
                "cover_url": track["album"]["images"][0]["url"],
                "spotify_url": track["external_urls"]["spotify"],
                "popularity": track["popularity"],
                "tempo": "fast" if track["tempo"] > 120 else "slow"
            })
        return SpotifySearchOutput(songs=songs)
    except Exception as e:
        log.error(f"Spotify 搜索失败：{e}")
        raise Exception(f"Spotify API 调用失败：{str(e)}")

# 注册工具
server = Server()
spotify_tool = Tool(
    name="spotify_search",
    description="通过 Spotify API 搜索音乐，支持关键词和流派筛选",
    input_schema=InputSchema(model=SpotifySearchInput),
    output_schema=OutputSchema(model=SpotifySearchOutput),
    handler=spotify_search
)
server.register_tool(spotify_tool)

# 启动 MCP Server
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(server.fastapi_app, host="0.0.0.0", port=8001)

步骤 2：Agent 调用 Spotify MCP 工具

修改 tools/music_tools.py，添加 MCP 客户端调用：

def search_music(state: MusicAgentState) -> MusicAgentState:
    """调用 Spotify MCP 工具搜索音乐"""
    # 解析搜索参数
    user_input = state.messages[-1].content
    llm = SiliconFlowLLM(**settings)
    params = llm.chat_completion([{"role": "user", "content": f"提取搜索关键词和流派：{user_input}"}])
    # 调用 MCP Server
    mcp_client = Client(base_url="http://localhost:8001")
    try:
        response = mcp_client.call_tool("spotify_search", {
            "keyword": params["keyword"],
            "genre": params.get("genre"),
            "limit": settings["MAX_RECOMMEND_NUM"]
        })
        state.recommendations = response.content[0].text["songs"]
    except Exception as e:
        state.error = f"搜索失败：{str(e)}"
    return state

步骤 3：前端添加 Spotify 播放功能

修改 music_app.py，支持点击播放和歌单创建：

def render_spotify_features(song):
    """渲染 Spotify 专属功能"""
    col1, col2 = st.columns(2)
    with col1:
        st.link_button("立即播放", song["spotify_url"], type="primary")
    with col2:
        if st.button("添加到歌单"):
            # 调用 Spotify API 创建歌单
            try:
                sp = init_spotify_client()
                playlist = sp.user_playlist_create(
                    user=sp.current_user()["id"],
                    name="AI 推荐歌单",
                    public=False
                )
                sp.playlist_add_items(playlist["id"], [song["spotify_url"].split("/")[-1]])
                st.success("已添加到歌单！")
            except Exception as e:
                st.error(f"添加失败：{str(e)}")

六、性能优化与生产级部署

1. 性能优化技巧

（1）缓存策略：减少重复调用

对相同意图和参数的请求，缓存推荐结果（使用 Redis 或本地缓存）：

def add_cache_middleware(func):
    """缓存装饰器"""
    cache = {}
    @wraps(func)
    def wrapper(state: MusicAgentState):
        # 生成缓存键（用户输入 + 意图）
        cache_key = f"{state.messages[-1].content}_{state.intent}"
        if cache_key in cache and time.time() - cache[cache_key]["timestamp"] < settings["CACHE_EXPIRE_SECONDS"]:
            state.recommendations = cache[cache_key]["data"]
            return state
        # 无缓存则执行原函数
        result = func(state)
        # 存入缓存
        cache[cache_key] = {
            "data": state.recommendations,
            "timestamp": time.time()
        }
        return result
    return wrapper

# 为推荐节点添加缓存
@add_cache_middleware
def recommend_by_activity(state: MusicAgentState) -> MusicAgentState:
    # 原有逻辑不变
    pass

（2）模型调优：平衡速度与效果

• 降低模型参数量：测试环境用 Qwen2.5-7B，生产环境根据预算选择 14B/72B；
• 调整温度参数：推荐场景用 0.2-0.3，避免推荐结果波动；
• 限制上下文长度：只保留最近 3 轮对话，减少 LLM 推理负担。

（3）异步处理：提升并发能力

使用 asyncio 异步调用 LLM 和 MCP 工具，支持多用户同时请求：

async def async_recommend_by_mood(state: MusicAgentState) -> MusicAgentState:
    """异步推荐节点"""
    loop = asyncio.get_event_loop()
    # 异步调用 LLM
    llm_task = loop.run_in_executor(None, llm.chat_completion, messages)
    # 异步调用 MCP 工具
    mcp_task = loop.run_in_executor(None, mcp_client.call_tool, "music_search", params)
    # 等待结果
    llm_result, mcp_result = await asyncio.gather(llm_task, mcp_task)
    # 处理结果
    state.recommendations = mcp_result.content[0].text["songs"]
    return state

2. 生产级部署方案

（1）Docker 容器化部署

创建 Dockerfile 和 docker-compose.yml，简化部署流程：

# Dockerfile
FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple
COPY . .
EXPOSE 8501
CMD ["python", "run_music_app.py"]

# docker-compose.yml
version: "3"
services:
  music-agent:
    build: .
    ports:
      - "8501:8501"
    environment:
      - SILICONFLOW_API_KEY=your_key
      - SPOTIFY_CLIENT_ID=your_id
      - SPOTIFY_CLIENT_SECRET=your_secret
    volumes:
      - ./data:/app/data
    restart: always
  mcp-server:
    build: ./mcp
    ports:
      - "8001:8001"
    restart: always

（2）云服务部署（以阿里云为例）

1. 创建 ECS 实例（推荐 2C4G 及以上）；
2. 安装 Docker 和 Docker Compose；
3. 克隆项目，配置 setting.json；
4. 执行 docker-compose up -d 启动服务；
5. 配置安全组，开放 8501 端口；
6. 访问 http://ECS公网IP:8501 即可使用。

（3）监控与日志

添加 Prometheus + Grafana 监控，或使用 Streamlit 的内置日志功能：

def setup_logging():
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
        handlers=[
            logging.FileHandler("logs/music_agent.log"),
            logging.StreamHandler()
        ]
    )

七、扩展方向与未来规划

1. 功能扩展

• 多模态推荐：结合歌词情感分析、封面图风格识别，提升推荐精准度；
• 用户偏好学习：存储用户历史反馈（喜欢/不喜欢），用向量数据库（如 Pinecone）构建个性化模型；
• 多平台对接：除 Spotify 外，支持网易云音乐、QQ 音乐 API；
• 语音交互：集成 Whisper 实现语音输入，支持“语音说需求，语音听推荐”。

2. 技术升级

• 多 Agent 协同：拆分意图识别、推荐、结果生成到不同 Agent，提升灵活性；
• 模型微调：用音乐推荐数据集微调 LLM，提升意图识别和推荐理由生成的准确性；
• 边缘部署：适配嵌入式设备（如智能音箱），支持离线推荐。

八、总结：智能推荐 Agent 的核心价值与落地建议

基于 LangGraph + MCP 的智能音乐推荐 Agent，核心价值在于“用自然语言理解打破推荐壁垒，用标准化工具调用实现生态扩展”。它不仅是一个 Demo 级项目，更是一套可复用的 Agent 构建框架——你可以将其适配到其他推荐场景（如电影、书籍、商品），只需替换工具层和数据层。

落地建议：

1. 初期用本地数据验证流程，再对接外部 API；
2. 优先优化意图识别和推荐逻辑，再扩展 MCP 工具和多平台集成；
3. 小流量测试后，逐步添加缓存、监控和部署优化；
4. 关注用户反馈，通过“反馈闭环”持续优化推荐效果。

除非注明，否则均为李锋镝的博客原创文章，转载必须以链接形式标明本文链接

本文链接：https://www.lifengdi.com/ren-gong-zhi-neng/4569