全栈 AI Agent 从 0 到 1 ：智能播客平台开发全记录

作者：蔡欣彤，京东科技
来源：京东技术
原文：https://mp.weixin.qq.com/s/IGSAgptmQ-ZDurI-pnhdog

一、引言：这个项目是做什么的？

Smart Podcast Platform 是一个端到端的智能播客制作平台。它的核心能力很简单：用户上传一段视频或音频，AI 自动理解内容、设计音色、生成播客，全程无需人工干预。

传统的 AI 音频工具通常停留在”生成文字脚本”的层面——用户拿到脚本后，还需要自己找配音、做剪辑、调音效，整个流程割裂且低效。这个项目试图把整个链条打通：从内容理解到音色设计，从语音合成到专业混音，全部由 AI Agent 自动完成。

这是我个人独立开发的全栈项目，从后端 Python 到前端 Vue 3，从 LangGraph Agent 编排到 pydub 音频处理，完整覆盖了一个 AI 音频应用的方方面面。

二、项目架构总览

2.1 技术栈一览

层级	技术选型	核心作用
前端框架	Vue 3+TypeScript+Vite	提供流式对话界面与播客管理
UI 组件	shadcn-vue+Tailwind CSS+ai-elements-vue	深色主题的现代化交互体验
AI 框架	LangChain1.0+LangGraph1.0.3	Agent 编排与多轮对话记忆
后端服务	FastAPI0.104.1+Uvicorn	REST API + SSE 流式响应
通信协议	AG-UI Protocol 0.1.18	前后端 Agent 事件标准化通信
语音引擎	阿里云DashScope (Qwen-TTS)	音色设计、语音克隆、语音合成
语音识别	阿里云DashScope (Qwen-ASR)	高精度语音转文字，支持 ITN 标准化
多模态模型	qwen3.5-omni-plus	图片/视频/音频统一理解
音频处理	pydub + FFmpeg	音频拼接、混音、后期制作

2.2 系统架构图

三、第一阶段：项目初始化与核心框架搭建

3.1 前端框架：Vue 3 + TypeScript

Vue 3使用Composition API（组合式API），逻辑聚合度更高，相比Vue 2的 Options API 更适合复杂交互场景。TypeScript 提供静态类型检查，帮我在编写阶段就发现大量潜在的 bug。

构建工具Vite冷启动极快，HMR（热模块替换）几乎感知不到延迟，开发体验远超Webpack。

3.2 UI组件：Tailwind CSS + shadcn-vue + ai-elements-vue

Tailwind CSS 采用原子化 CSS 方案，不需要单独维护 CSS 文件，所有样式直接写在类名上。下面是音频卡片组件的样式实现，选中和悬浮状态完全用 Tailwind 类名控制：

<!-- AudioCard.vue 音频卡片：选中态/悬浮态用 Tailwind 条件类名实现 -->
<div
  class="group relative rounded-lg border p-3 transition-all duration-200 cursor-pointer"
  :class="selected
    ? 'border-cyan-500/50 bg-cyan-500/[0.08] shadow-[0_0_12px_rgba(34,211,238,0.1)]'
    : 'border-cyan-500/10 bg-[#111827]/60 hover:border-cyan-500/30 hover:bg-cyan-500/[0.05]'"
  @click="emit('select')"
>
  <!-- 选中指示条 -->
  <div v-if="selected" class="absolute left-0 top-3 bottom-3 w-0.5 rounded-full bg-cyan-400" />
  <!-- 音频播放器 -->

    <audio controls class="w-full h-8 rounded opacity-80 hover:opacity-100 transition-opacity" :src="item.url" />
</div>

hadcn-vue 是基于 Reka UI 的组件库，它的特点是组件代码直接复制到项目里，完全可定制，而不是黑盒的 npm 包。Button、Dialog、Input 这些基础组件拿来即用。

ai-elements-vue 是专门为 AI 对话场景设计的组件库，项目里用到了：

• Conversation/ConversationContent：对话容器，自动处理滚动
• Message/MessageContent/MessageResponse：消息气泡，支持Markdown渲染
• PromptInput/PromptInputTextarea：输入框，内置提交状态管理
• ConversationScrollButton：自动吸底滚动按钮

这些组件让我省去了几乎所有 AI 对话 UI 的基础建设，专注在业务逻辑上。

<Conversation class="h-full">
  <ConversationContent>
    <!-- Empty State -->
    <ConversationEmptyState
      v-if="messages.length === 0"
      title="开始音频对话"
      description="输入内容，与 AI 音频智能体开始交流"
    />

        <!-- Messages -->
    <template v-else>
      <Message v-for="(message, index) in messages" :key="index" :from="message.role">
        <div class="flex items-start gap-3">
          <MessageAvatar
            v-if="message.role === 'assistant'"
            src="/ai-avatar.png"
            name="AI"
          />
          <MessageAvatar v-else src="/user-avatar.png" name="用户" />
          <MessageContent>
            <MessageResponse :content="message.content" />
          </MessageContent>
        </div>
      </Message>
    </template>
  </ConversationContent>
</Conversation>

3.3 后端框架：FastAPI

FastAPI是Python生态中性能优秀的异步Web框架，基于ASGI标准，天然支持流式响应。

# main.py 入口：注册路由 + 挂载静态文件（音频直链访问）
app.include_router(chat.router, prefix="/api", tags=["chat"])
app.include_router(resources.router, prefix="/resources", tags=["resources"])
app.mount("/storage", StaticFiles(directory=storage_path), name="storage")
# routers/chat.py 聊天接口：直接返回 StreamingResponse
@router.post("/chat")
async def chat_normal(request: Request, chat_request: ChatRequest):
    accept_header = request.headers.get("accept", "text/event-stream")
    encoder = EventEncoder(accept=accept_header)
    return StreamingResponse(
        process_agent_stream(chat_request.message, chat_request.thread_id, encoder),
        media_type=encoder.get_content_type(),
    )

3.4 AI Agent核心：LangChain 1.0 + LangGraph

这是整个项目技术含量最高的部分，也是我做了最多功能抽离和设计的地方。

• LangChain 1.0 全新 API

项目使用的是 LangChain 1.0.0，这个版本相比旧版有较大 API 变动，很多网上的教程代码已经无法直接用。核心变化是 Agent 创建方式统一为 create_agent，工具注册更加简洁。

• LLM 工厂模式（factory.py）

我把 LLM 实例的创建单独抽成了一个工厂函数，而不是在 Agent 里直接 hard-code。好处是以后切换模型（比如从 DeepSeek 换成 Qwen）只需要改环境变量，不需要动业务代码：

# app/llm/factory.py — LLM 工厂，所有配置从环境变量读取
def create_llm(temperature: float = 0.7, max_tokens=None, **kwargs) -> ChatOpenAI:
    openai_api_key = os.getenv("OPENAI_API_KEY")
    base_url = os.getenv("OPENAI_API_BASE")
    model_name = os.getenv("MODEL_NAME", "deepseek-chat")
    return ChatOpenAI(
        model=model_name,
        api_key=openai_api_key,
        base_url=base_url,
        temperature=temperature,
        max_tokens=max_tokens,
        **kwargs
    )

• Prompt 模块抽离（prompt.py）

系统提示词单独放在 services/prompt.py 里，不和 Agent 初始化逻辑混在一起。而且 Prompt 支持动态注入工具列表描述，Agent 初始化时会自动把已注册的工具名称和描述拼入Prompt，避免提示词和代码不一致：

# agent_service.py — 动态生成工具列表，注入 Prompt
tool_descriptions = []
for tool in tools:
    description = getattr(tool, 'description', None)
    tool_descriptions.append(f"- {tool.name}: {description}")
tools_list_text = "
".join(tool_descriptions)
full_prompt = get_full_prompt(tools_list_text)  # 注入到系统提示词

• Agent创建与工具注册

# agent_service.py — Agent 创建，工具注册，InMemorySaver 持久化多轮记忆
def create_multimodal_agent():
    model = create_llm(temperature=0.7)
    tools = [
        qwen_voice_design_tool,   # 工具1：文字描述生成定制语音
        qwen_voice_cloning_tool,  # 工具2：音色复刻 + 语音合成
    ]
    agent = create_agent(
        name="tts_agent",
        model=model,
        tools=tools,
        system_prompt=full_prompt,
        checkpointer=InMemorySaver()  # 多轮对话记忆，按 thread_id 隔离
    )
    return agent
# 模块加载时初始化一次，全局复用
agent = create_multimodal_agent()

• LangGraph的多轮记忆机制

LangGraph 的 InMemorySaver 会按 thread_id 保存每次对话的完整消息历史。每次用户发新消息，只需要传入当前这条，LangGraph 会自动从 checkpoint 中恢复上下文：

注意：InMemorySaver 只适合本地简单尝试，真实业务需要用数据库

# process_agent_stream — 每次只传当前消息，历史由 LangGraph 自动管理
async def process_agent_stream(message: str, thread_id: str = "default", encoder=None):
    processor = StreamProcessor(thread_id, encoder=encoder)
    messages = [HumanMessage(content=message)]  # 只传当前消息
    async for event in processor.process_stream(agent, messages):
        yield event

这样的设计好处是：前端不需要维护对话历史、不需要每次把全量历史发给后端，后端按 thread_id 自动恢复，接口保持简洁。

3.5 流式通信：SSE + AG-UI 协议

这是项目的通信层核心，实现了前后端之间结构化、实时的事件流交互。

• 为什么用 SSE而不是WebSocket？

SSE（Server-Sent Events）是单向的服务器推送，基于普通 HTTP 连接，比 WebSocket 轻量得多。对于 AI 对话这种”用户发一条，AI 持续回复”的场景，SSE 完全够用，而且不需要额外的握手和连接管理。

• AG-UI协议是什么？

AG-UI 是一套专门为 AI Agent 与前端通信设计的事件规范，定义了标准的事件类型：

事件类型	含义
RUN_STARTED	Agent开始运行
TEXT_MESSAGE_START	文本消息开始
TEXT_MESSAGE_CONTENT	文本增量内容（流式输出每一块）
TEXT_MESSAGE_END	文本消息结束
TOOL_CALL_START	开始调用工具
TOOL_CALL_ARGS	工具调用参数
TOOL_CALL_END	工具调用结束
TOOL_CALL_RESULT	工具调用结果（含音频 URL）
RUN_FINISHED	Agent运行完成

• 后端：StreamProcessor事件分发

我把SSE事件的编码和分发封装成了独立的StreamProcessor类，和Agent逻辑完全解耦：

# stream_processor.py — 核心流式处理逻辑
class StreamProcessor:
    def __init__(self, thread_id: str, encoder=None):
        self.thread_id = thread_id
        self.encoder = encoder or EventEncoder(accept="text/event-stream")
    async def _handle_chunk(self, chunk):
        """处理每个 chunk，按消息类型分发对应的 AG-UI 事件"""
        message_chunk = chunk[0] if isinstance(chunk, tuple) else chunk
        if isinstance(message_chunk, AIMessage):
            if message_chunk.tool_calls:
                # 工具调用：发送 TOOL_CALL_START → TOOL_CALL_ARGS → TOOL_CALL_END
                for tool_call in message_chunk.tool_calls:
                    tool_call_id = f"tool_{self.thread_id}_{tool_call['name']}"
                    yield self.encoder.encode(ToolCallStartEvent(
                        type=EventType.TOOL_CALL_START,
                        tool_call_id=tool_call_id,
                        tool_call_name=tool_call["name"],
                        parent_message_id=f"msg_{self.thread_id}"
                    ))
                    yield self.encoder.encode(ToolCallArgsEvent(
                        type=EventType.TOOL_CALL_ARGS,
                        tool_call_id=tool_call_id,
                        delta=json.dumps(tool_call["args"], ensure_ascii=False)
                    ))
                    yield self.encoder.encode(ToolCallEndEvent(
                        type=EventType.TOOL_CALL_END,
                        tool_call_id=tool_call_id
                    ))
            else:
                # 普通文本：发送 TEXT_MESSAGE_CONTENT（逐字符）
                if message_chunk.content:
                    yield self.encoder.encode(TextMessageContentEvent(
                        type=EventType.TEXT_MESSAGE_CONTENT,
                        messageId=f"msg_{self.thread_id}",
                        delta=message_chunk.content
                    ))
        elif isinstance(message_chunk, ToolMessage):
            # 工具执行完毕：发送 TOOL_CALL_RESULT（含音频 URL）
            yield self.encoder.encode(ToolCallResultEvent(
                type=EventType.TOOL_CALL_RESULT,
                tool_call_id=f"tool_{self.thread_id}_{message_chunk.name}",
                tool_name=message_chunk.name,
                content=message_chunk.content,
                role="tool"
            ))

• 前端：AG-UI 事件解析与 UI 更新

前端 chat.ts 中封装了 dispatchAGUIEvent，统一解析事件类型，把文本增量和工具结果分别回调给上层：

// api/chat.ts — 前端事件分发，解耦协议解析和 UI 更新
function dispatchAGUIEvent(event: AGUIEvent, onMessage: (chunk: SSEChunk) => void) {
  if (event.type === 'TEXT_MESSAGE_CONTENT' && event.delta) {
    onMessage({ type: 'text', data: event.delta })
  } else if (event.type === 'TOOL_CALL_RESULT' && event.content) {
    const toolResult: ToolCallResult = {
      toolCallId: event.tool_call_id || '',
      toolName: event.tool_call_name || '',
      content: JSON.parse(event.content),
    }
    onMessage({ type: 'TOOL_CALL_RESULT', data: JSON.stringify(toolResult) })
  }
}

ChatAgent.vue 在流式回调中按事件类型更新 UI：

// views/ChatAgent.vue — 流式回调，实时更新对话状态
await sendChatMessage(message.text, thread_id.value, currentMode.value,
  (chunk: SSEChunk) => {
    if (chunk.type === 'text') {
      // 打字机效果：逐块追加文本
      messages.value[assistantIndex].content += chunk.data
    } else if (chunk.type === 'TOOL_CALL_RESULT') {
      // 工具调用结果：展示音频播放器卡片
      const toolResult: ToolResult = JSON.parse(chunk.data)
      messages.value[assistantIndex].toolResults ??= []
      messages.value[assistantIndex].toolResults.push(toolResult)
    }
  }
)

整个通信链路如下：

用户发消息
  ↓ POST /api/chat（带 thread_id）
FastAPI 接收，创建 StreamingResponse
  ↓
LangGraph Agent 流式执行（stream_mode="messages"）
  ↓ 每个 chunk 经 StreamProcessor 转换
AG-UI 事件（SSE 格式推送到前端）
  ↓ 前端 dispatchAGUIEvent 解析
Vue 响应式更新 UI（文本打字机 / 工具结果卡片）

3.6 第一阶段踩过的坑

1. LangChain 1.0 版本变化大：旧版的 initialize_agent、AgentExecutor 等 API 全部废弃，新版统一用 create_agent，网上大部分教程代码无法直接用，建议直接看官方 1.0 Changelog。
2. SSE 响应被缓冲：后端加了 gzip 压缩中间件后，SSE 数据会被缓冲等凑满再发，导致前端收不到流式效果。解决方式是给 SSE 响应添加 X-Accel-Buffering: no 响应头。
3. 跨域 + 代理配置：前端 3000 端口、后端 8000 端口，需要同时配置 FastAPI 的 CORS 中间件和 Vite 的 proxy，缺一不可。
4. Python 虚拟环境：不同项目的依赖版本冲突是真实存在的问题，venv 或 conda 隔离是必须做的事。

3.7 第一阶段功能展示

• 音频创作对话效果

• 音色复刻功能

四、第二阶段：播客后期制作能力

4.1 音频混音工具模块

新增 backend/app/tools/audio_mixing.py，提供三个核心音频处理工具：

4.1.1 音频拼接工具（concatenate_audio）

将多个音频片段按顺序拼接成完整对话，适用于播客场景。

核心功能：

• 支持交叉淡入淡出过渡（crossfade），使音频过渡更自然
• 可配置音频片段之间的静音时长（默认1200ms，播客推荐1000-1500ms）
• 自动生成带时间戳的唯一文件名
• 自动记录到音频索引系统

@tool("concatenate_audio", args_schema=ConcatenateAudioInput)
def concatenate_audio_tool(
    audio_files: List[str],       # 音频文件路径列表
    crossfade_duration: int = 200,  # 交叉淡入淡出时长（ms）
    silence_duration: int = 1200    # 静音间隔（ms）
) -> str:
    """
    拼接逻辑：
    1. 加载所有音频片段 → AudioSegment.from_file()
    2. 依次拼接，支持两种模式：
       - crossfade > 0：交叉淡入淡出过渡
       - silence > 0：静音间隔拼接
    3. 导出为 WAV 文件并记录索引
    """

4.1.2 智能BGM选择工具（select_background_music）

根据场景描述智能匹配合适的背景音乐。

核心功能：

• 基于文件名的语义匹配（BGM文件名即场景描述）
• 自动循环播放或裁剪以匹配目标时长
• 支持直接指定BGM文件路径
• 自动添加淡出效果

@tool("select_background_music", args_schema=SelectBGMInput)
def select_bgm_tool(
    scene_description: str,           # 场景描述，如"欢快的开场"
    duration_seconds: Optional[float] = None,  # 期望时长
    bgm_audio_path: Optional[str] = None       # 指定BGM路径
) -> str:
    """
    匹配逻辑：
    1. 如果指定了 bgm_audio_path，直接使用
    2. 否则扫描 storage/bgm/ 目录下所有 .mp3/.wav 文件
    3. 基于文件名与 scene_description 的关键词匹配度打分
    4. 选择得分最高的BGM，必要时循环/裁剪匹配时长
    """

4.1.3 音频混音工具（mix_audio_with_bgm）

将人声对话与背景音乐混合，生成专业的播客成品。

核心功能：

• 专业BGM开场效果
  • 前N秒（默认3秒）：BGM原音量播放
  • 过渡阶段：音量渐变至背景音量
  • 后续阶段：BGM降低至约5%音量作为背景
• 音量归一化处理，确保音质一致性
• 自动淡出效果
• 支持自定义BGM音量（推荐-24到-28dB）

@tool("mix_audio_with_bgm", args_schema=MixAudioWithBGMInput)
def mix_audio_with_bgm_tool(
    voice_audio: str,           # 主音频路径（人声）
    bgm_audio: str,             # BGM路径
    bgm_volume: float = -26,    # BGM背景音量（dB）
    intro_duration: float = 3.0, # BGM开场时长（秒）
    normalize: bool = True       # 是否归一化
) -> str:
    """
    混音逻辑：
    1. 加载人声和BGM音频
    2. BGM分段处理：
       - 开场段：原音量播放 intro_duration 秒
       - 过渡段：20步渐变从原音量到 bgm_volume
       - 背景段：固定 bgm_volume 音量
    3. 人声前插入静音（与BGM开场对齐）
    4. 叠加人声到BGM → bgm_processed.overlay(voice_with_intro)
    5. 可选归一化 + 淡出，导出为 MP3
    """

音量变化曲线：

BGM 音量变化：
原音量 ─────┐
            │\
            │ \  过渡时段（2秒）
            │  \
背景音量 ───┘   └────────── 背景音量
         ↑
      开场时段（3秒）

4.2 音频资源管理API

新增 backend/app/routers/resources.py，提供音频资源的RESTful API。

方法	路径	说明
GET	/resources/audio	查询所有音频资源列表
POST	/resources/audio/upload	上传音频文件（支持 .mp3 和 .wav）
DELETE	/resources/audio/{id}	删除指定音频资源

上传接口代码示例

@router.post("/audio/upload", response_model=UploadResponse)
async def upload_audio_resource(file: UploadFile = File(...)):
    """上传音频文件，存储到 BGM_DIR 并记录到索引"""
    # 1. 校验文件格式（仅支持 .mp3 和 .wav）
    ext = os.path.splitext(file.filename)[1].lower()
    if ext not in UPLOAD_ALLOWED_EXTENSIONS:
        raise HTTPException(status_code=400, detail=f"不支持的文件格式")
    # 2. 生成唯一文件名，避免冲突
    voice_id = str(uuid.uuid4())
    safe_filename = f"{name_without_ext}{voice_id}{ext}"
    # 3. 保存文件到 storage/bgm/
    with open(save_path, "wb") as f:
        shutil.copyfileobj(file.file, f)
    # 4. 记录到音频索引
    record_voice_index(local_path=save_path, voice=voice_id, model_name="", path="bgm")
    return UploadResponse(success=True, message="上传成功", voice_id=voice_id)

删除接口代码示例

@router.delete("/audio/{id}", response_model=DeleteResponse)
async def delete_audio_resource(id: str):
    """删除音频资源：移除索引记录 + 删除物理文件"""
    # 1. 从 voice_index.json 中查找匹配记录
    matched = [r for r in voice_index if r.get("id") == id]
    # 2. 删除对应的本地文件
    for record in matched:
        filepath = os.path.abspath(record.get("local_path", ""))
        os.remove(filepath)
    # 3. 从索引中移除并写回
    voice_index = [r for r in voice_index if r.get("id") != id]
    with open(index_path, "w", encoding="utf-8") as f:
        json.dump(voice_index, f, ensure_ascii=False, indent=2)
    return DeleteResponse(success=True, message="删除成功")

4.3 音频索引系统

新增 backend/app/tools/audio_index.py，提供统一的音频索引管理，供各工具模块共享调用。

def record_voice_index(local_path: str, voice: str, model_name: str = "", path: str = "audios") -> None:
    """将音频文件信息记录到 voice_index.json 索引文件中
    Args:
        local_path: 音频文件本地路径
        voice: 音色ID
        model_name: 模型名称
        path: 音频存储路径类型 (audios/bgm/podcasts)
    """
    # 读取已有索引 → 追加新记录 → 写回文件
    new_record = {
        "id": str(uuid.uuid4()),
        "local_path": local_path,
        "voice_id": voice,
        "model_name": model_name,
        "path": path,
        "createTime": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    }
    index.append(new_record)
    VOICE_INDEX_FILE.write_text(json.dumps(index, ensure_ascii=False, indent=2))

path字段说明：

path值	说明	存储目录
audios	TTS生成的音频	storage/audios/
bgm	上传的背景音乐	storage/bgm/
podcasts	混音生成的播客成品	storage/podcasts/

4.4 储存目录结构

storage/
├── audios/              # TTS生成的音频
├── bgm/                 # 背景音乐（新增）
├── podcasts/            # 播客成品（新增）
├── voice_index.json     # 音频索引（统一管理）
└── images/              # 图片文件

4.5 前端更新：语音输入转文字

在聊天输入框中新增语音输入功能，用户可以通过麦克风将语音转换为文字发送消息。

技术实现：

• 基于 Web Speech API（SpeechRecognition / webkitSpeechRecognition）实现浏览器端语音识别
• 默认语言为中文（zh-CN），支持通过 lang 属性配置
• 支持连续识别（continuous: true）和中间结果（interimResults: true）
• 识别完成后自动将文字填入输入框，智能处理中英文空格

// 初始化语音识别
const sr = new SpeechRecognition()
sr.continuous = true          // 连续识别
sr.interimResults = true      // 返回中间结果
sr.lang = 'zh-CN'             // 默认中文
// 识别结果回调
sr.onresult = (event) => {
  let finalTranscript = ''
  for (let i = event.resultIndex; i < event.results.length; i++) {
    if (event.results[i].isFinal) {
      finalTranscript += event.results[i][0]?.transcript ?? ''
    }
  }
  if (finalTranscript) {
    // 智能添加空格：中文不加空格，英文前加空格
    const needsSpace = textInput.value && !/[\u4e00-\u9fa5]/.test(textInput.value.slice(-1))
    setTextInput(textInput.value + (needsSpace ? ' ' : '') + finalTranscript)
  }
}

交互效果：

• 点击麦克风按钮开始录音，按钮显示脉冲动画
• 再次点击停止录音
• 识别结果自动追加到输入框文本中
• 不支持 Web Speech API 的浏览器自动禁用按钮

4.6 第二阶段功能展示

五、第三阶段：多模态识别与储存架构升级

5.1 音频、视频文件储存路径重构

变更文件：backend/app/tools/audio_index.py

原先音频文件统一保存在 storage/audios/ 目录下，现在改为保存到临时目录 storage/temp/，实现临时文件与永久文件的分离管理。

核心改动：新增 save_media_to_temp() 函数，支持三种输入类型：

def save_media_to_temp(media_data, filename: str) -> str:
    """将音频/视频文件保存到临时目录"""
    temp_dir = STORAGE_DIR / "temp"
    temp_dir.mkdir(parents=True, exist_ok=True)
    file_path = temp_dir / filename
    if isinstance(media_data, str):
        # base64 字符串：解码保存
        file_data = base64.b64decode(media_data)
        file_path.write_bytes(file_data)
    elif isinstance(media_data, bytes):
        # bytes 对象：直接保存
        file_path.write_bytes(media_data)
    else:
        # AudioSegment 对象：导出文件
        file_format = filename.split('.')[-1]
        media_data.export(str(file_path), format=file_format)
    return f"storage/temp/{filename}"

设计思路：音色设计、声音克隆等工具生成的中间产物统一放在 storage/temp/，当用户确认满意后再通过 save_voice 工具将音色永久迁移到 storage/audios/。这样既避免了永久目录的膨胀，也让文件生命周期管理更清晰。

5.2 定时临时文件清理任务

变更文件：backend/app/utils/temp_cleanup.py（新增）新增定时清理机制，自动清除 storage/temp/ 目录下超过指定时间的过期文件，防止临时文件堆积占用磁盘空间。

def cleanup_temp_files(max_age_minutes: int = 10) -> dict:
    """清理过期的临时文件"""
    cutoff_time = time.time() - (max_age_minutes * 60)
    for file_path in TEMP_DIR.rglob("*"):
        if not file_path.is_file():
            continue
        file_mtime = file_path.stat().st_mtime
        if file_mtime < cutoff_time:
            file_size = file_path.stat().st_size
            file_path.unlink()
            stats["deleted_files"] += 1
            stats["freed_bytes"] += file_size
    # 清理空目录
    cleanup_empty_dirs(TEMP_DIR)
    return stats

关键特性：

• 默认保留时间10分钟，可灵活配置
• 递归清理所有子目录，并自动移除空目录
• 返回详细统计信息（文件数、删除数、释放空间）
• 提供 schedule_cleanup_task() 函数，方便接入 APScheduler 等定时调度框架

5.3 音色保存工具

变更文件：backend/app/tools/voice_save.py（新增）

新增save_voice 工具，将用户满意的定制音色从临时目录永久保存到storage/audios/，并记录到voice_index.json索引文件。

@tool("save_voice", args_schema=VoiceSaveInput)
def save_voice_tool(audio_source: str, voice_id: str, text: str, model_name: str = "") -> str:
    # 判断输入类型：文件路径 或 base64数据
    if audio_source.startswith("storage/") or audio_source.startswith("/"):
        # 从临时目录复制到永久目录
        source_path = BASE_DIR / audio_source.lstrip("/")
        dest_path = AUDIOS_DIR / source_path.name
        shutil.copy2(source_path, dest_path)
        local_path = f"storage/audios/{source_path.name}"
    else:
        # base64 数据：解码并保存
        local_path = save_audio_from_base64(audio_source, text, "saved_voice")
    # 记录到索引文件
    record_voice_index(local_path, voice_id, model_name, path="audios")
    return json.dumps({"audio_url": local_path, "voice_id": voice_id, ...})

设计亮点：

• 支持两种输入方式：临时文件路径（直接复制）和 base64 编码数据（解码保存）
• 自动生成唯一文件名，包含时间戳、UUID、音色ID和文本片段
• 使用文件锁（fcntl.flock）保证索引文件并发写入安全
• 与播客工作流无缝衔接：音色设计 → 临时保存 → 用户确认 → 永久保存

5.4 语音识别工具（ASR）

变更文件：backend/app/tools/qwen_asr.py（新增）

集成阿里云 DashScope qwen3-asr-flash 模型，提供高精度语音识别能力。

@tool("qwen_asr_tool", args_schema=ASRInput)
def qwen_asr_tool(audio: str, enable_itn: bool = False) -> str:
    # 1. 解析音频来源（URL / 本地路径 → data URI）
    audio_url = _resolve_audio_source(audio)
    # 2. 调用 ASR API
    response = _call_asr_api(audio_url, enable_itn)
    # 3. 提取识别文本
    text = _extract_text_from_response(response)
    return text

关键能力：

• 多格式支持：MP3、WAV、OGG、FLAC、M4A、AAC 等主流音频格式
• ITN 逆文本标准化：自动将口语化表达转为书面形式（如”二零二五年” → “2025年”）
• 智能来源解析：支持本地路径和网络 URL，本地文件自动转为 base64 data URI
• MIME 类型自动推断：根据文件扩展名匹配正确的 MIME 类型

5.5 多模态识别工具

变更文件：backend/app/tools/qwen_multimodal.py（新增）

集成阿里云 DashScope qwen3.5-omni-plus 模型，支持图片、视频、音频的统一多模态理解。

大视频智能分割：当视频文件超过 21MB 时，自动使用 moviepy 分割为多个片段分别处理：

def _split_and_encode_video(video_path: Path, suffix: str) -> List[dict]:
    video = VideoFileClip(str(video_path))
    target_size = 10 * 1024 * 1024  # 每段 10MB
    num_segments = max(1, int(file_size / target_size))
    segment_duration = duration / num_segments
    for i in range(num_segments):
        segment = video.subclipped(start_time, end_time)
        # 编码并添加到片段列表
        segments.append(_build_media_content(data_uri, suffix, is_url=False))
    return segments

额外能力：

• 支持流式输出（stream=True），实时获取识别结果
• 支持音频输出模态（enable_audio_output），可指定音色进行语音合成
• 完善的错误处理和日志追踪

5.6 提示词优化

变更文件：backend/app/services/prompt.py

对播客 Agent 的系统提示词进行了全面精炼和增强：

优化要点：

• 结构重组：将原本杂乱的提示词拆分为清晰的模块化结构——核心能力定位、工作方式、工具调用规则、沟通规范、上下文理解、执行流程
• 新增音视频识别内容：在工作流中明确区分语音识别（qwen_asr_tool）和多模态理解（qwen_multimodal_tool / qwen_combined_multimodal_tool）的使用场景
• 工具使用策略表格化：用清晰的映射关系说明各工具职责，降低 LLM 误调用概率音色
• 保存流程标准化：明确”设计 → 临时保存 → 用户确认 → 永久保存”的四步流程
• 避免重复调用：在提示词中强调”对于简单请求，调用一次工具后立即结束回复”

新增的音视频识别工作流描述：

**音视频转播客工作流**：
1. 内容识别：调用 qwen_multimodal_tool 或 qwen_asr_tool 识别音视频中的内容
2. 脚本整理：基于识别出的内容，整理为播客脚本
3. 播客制作：按照工作流步骤继续执行，完成播客制作

5.7 架构总结

本次更新的核心脉络：

用户上传音视频
    │
    ▼
┌─────────────────────────────────────────────┐
│  内容识别层                                  │
│  ├─ qwen_asr_tool         语音→文字         │
│  └─ qwen_multimodal_tool  音视频→理解       │
└─────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────┐
│  播客制作层                                  │
│  ├─ 音色设计 → 临时目录 (storage/temp/)     │
│  ├─ save_voice → 永久目录 (storage/audios/) │
│  └─ 音频合成 / 混音 / 拼接                   │
└─────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────┐
│  系统维护层                              │
│  └─ temp_cleanup  定时清理临时文件      │
└─────────────────────────────────────────────┘

通过”临时-永久”双层存储架构 + 定时清理机制，既保证了播客制作流程的灵活性，也确保了系统资源的有效管理。

六、第四阶段：配置管理，资源管理与工程实践总结

6.1 配置管理：多层级优先级策略

配置文件 > 环境变量 > 默认值的三级配置优先级：

# backend/app/utils/config_manager.py
def get_config_value(key: str, env_key: str, default: str = "") -> str:
    # 1. 先读 config.json（前端可视化配置）
    config = load_config()
    value = config.get("llm", {}).get("openai_api_key")
    # 2. 再读环境变量
    if not value:
        value = os.getenv("OPENAI_API_KEY")
    # 3. 最后用默认值
    return value or default

设计意图：用户可以通过前端界面（VisualConfig.vue）直接修改配置，无需手动编辑 .env 文件。配置自动持久化到 storage/config.json，降低使用门槛。

6.2 音频索引系统：轻量级的资源管理

我们没有引入数据库，而是用 JSON 文件实现了轻量级的音频资源索引：

// storage/voice_index.json
[
  {
    "id": "uuid",
    "local_path": "storage/audios/xxx.wav",
    "voice_id": "voice-xxx",
    "model_name": "cosyvoice-v3.5-plus",
    "path": "audios",
    "createTime": "2026-07-10 14:30:00"
  }
]

关键设计：

• 文件锁并发控制：使用 fcntl.flock 保证多请求并发写入安全
• path字段分类：audios（定制音色）/ bgm（背景音乐）/ podcasts（播客成品），便于前端分类展示
• 临时文件自动清理：storage/temp/ 目录下的文件定期清理，避免磁盘堆积

6.3 前端交互设计

6.3.1 深色主题的沉浸式体验

前端采用品牌深蓝紫配色 (Brand Deep Blue-Purple Palette),深空背景 + 蓝紫光晕 + 亮蓝高亮：

:root {
  /* —— 背景渐变（页面主底色，从深空蓝到深紫）—— */
  --brand-bg-from: #050818;
  --brand-bg-via:  #0a0f2e;
  --brand-bg-to:   #0d0524;
  /* —— 光晕装饰色 —— */
  --brand-glow-blue:      #3a5cff;
  --brand-glow-purple:    #7c3aed;
  --brand-glow-deep-blue: #1e40ff;
  /* —— 品牌高亮蓝 —— */
  --brand-blue:        #4f7cff;
  --brand-blue-light:  #4f9dff;
  --brand-blue-strong: #2b6ef7;
  --brand-blue-soft:   #7ea3ff;
  --brand-blue-pale:   #9ec2ff;
  /* —— 卡片封面渐变 —— */
  --brand-cover-from: #1e3a8a;
  --brand-cover-via:  #2a1e5f;
  --brand-cover-to:   #0d0524;
  /* —— Header 装饰渐变 —— */
  --brand-header-from: #1e3a8a;
  --brand-header-via:  #3b2a8a;
}

6.3.2 流式对话体验

基于 ai-elements-vue 组件库，前端实现了丰富的 Agent 交互界面：

• 思维链展示：Agent 的推理步骤以可折叠卡片形式展示
• 工具调用可视化：每个工具调用显示名称、参数、结果
• 音频播放器：生成的音频直接在对话中嵌入播放
• 文件上传：支持拖拽上传音频/视频文件

6.3.3 四个核心页面

七、工程实践与经验

7.1 工具设计的”单一职责”原则

我们将音频处理拆分为 9 个独立工具，每个工具只做一件事：

工具	职责	输入	输出
qwen_multimodal_tool	多模态内容理解	媒体文件	文本描述
qwen_asr_tool	语音识别	音频文件	文字转录
qwen_voice_design	音色设计	文本描述	音色ID+音频
qwen_voice_cloning	语音合成	音色ID+文本	音频文件
save_voice	音色保存	音频路径	永久存储
concatenate_audio	音频拼接	音频列表	拼接文件
select_background_music	BGM 选择	场景描述	BGM 路径
mix_audio_with_bgm	专业混音	人声+BGM	最终播客

这种设计让 Agent 可以灵活组合工具，而不是被预设的”大而全”工具束缚。

7.2 临时文件与永久存储的分离

• 临时目录（storage/temp/）：工具生成的中间产物，10 分钟自动清理
• 永久目录（storage/audios/）：用户确认保存的音色，持久化存储
• 成品目录（storage/podcasts/）：混音后的最终播客

这个设计避免了磁盘空间浪费，同时保证了用户主动保存的内容不会丢失。

7.3 流式输出的”真”流式

我们使用 stream_mode=”messages” 模式，确保 Agent 的输出是逐 token 流式的，而非”生成完再发送”的伪流式：

async for chunk in agent.astream(
    {"messages": langchain_messages},
    {"configurable": {"thread_id": self.thread_id}},
    stream_mode="messages"  # 关键：逐消息流式输出
):
    async for event in self._handle_chunk(chunk):
        yield event  # 立即发送，不等待

7.4 配置可视化：降低技术门槛

传统 AI 项目需要用户手动编辑 .env 文件，对非技术用户极不友好。我们通过 VisualConfig.vue 页面，将配置项以表单形式呈现：

• API Key 输入框（支持密码可见性切换）
• 模型选择下拉框
• 配置自动持久化到 config.json
• 前后端通过 /api/config 接口同步

八、总结

这个项目展示了 AI Agent 在垂直内容创作领域的完整实践：

1. 端到端自动化：从内容理解到播客输出，全流程 AI 驱动
2. 多模态融合：文字、图片、音频、视频统一处理
3. 专业级音频质量：音色设计 + 智能混音，输出广播级品质
4. 低门槛使用：可视化配置 + 自然语言交互，无需技术背景
   对于同样在学习 AI 应用开发的朋友，有几点心得：
5. 框架版本要锁定：LangChain 这类快速迭代的框架，版本差异带来的问题远比你想象的多
6. 先跑通链路，再追求完美：新手阶段能把用户输入 → AI 处理 → 结果展示这条链跑通，比深入某个技术点更有价值
7. 多看官方文档：相比博客和教程，官方文档更新及时，特别是 LangGraph 这种新框架
8. 工具设计要单一职责：每个工具只做一件事，让 Agent 灵活组合而非被预设束缚
9. 存储要分层管理：临时-永久双层架构，避免磁盘膨胀同时保证数据安全

九、附录：快速开始

# 1. 克隆项目
git clone 
# 2. 启动后端
cd backend
python -m venv venv
source venv/bin/activate
pip install -r requirements.txt
cp .env.example .env  # 填写 API Key
python main.py
# 3. 启动前端
cd fronted
npm install
npm run dev
# 4. 访问 http://localhost:3000