【全栈实践】第一个 AI Agent 项目：从零搭建 AI 音频创作助手-京东云开发者社区

## 项目是什么？

这是一个 **AI 驱动的音频内容创作助手**。你可以用自然语言与它对话，它会理解意图、调用相应的语音合成工具，把文字变成真实的音频文件，并保存在本地。

### 核心功能

* **文字描述生成语音**：告诉 AI"帮我生成一段温柔女声播报618大促的语音"，它会根据音色描述调用语音设计工具，直接生成定制化音频
* **音色复刻合成语音**：放入参考的音频地址，AI 会复刻该音色，用同样的嗓音合成新的文字内容
* **已复刻音色直接合成**：对已经保存的复刻音色，可以直接指定音色名称进行语音合成，无需重复复刻音色
* **本地音频资源展示**：所有生成的音频文件会保存在本地，侧边栏实时展示音频列表，支持在线播放、复制链接、管理文件

整个项目分为两部分：

* **前端**：负责对话界面、流式消息展示、工具调用卡片、音频播放
* **后端**：负责接收消息、驱动 AI Agent、调用 TTS 工具、管理音频文件

***

## 技术栈总览

### 前端

| 技术 | 版本 | 文档 |
| --- | --- | --- |
| Vue 3 | ^3.5.31 | [vuejs.org](https://vuejs.org/) |
| TypeScript | \~6.0.0 | [typescriptlang.org](https://www.typescriptlang.org/) |
| Vite | ^8.0.3 | [vite.dev](https://vite.dev/) |
| Tailwind CSS | ^4.2.2 | [tailwindcss.com](https://tailwindcss.com/) |
| shadcn-vue | - | [shadcn-vue.com](https://www.shadcn-vue.com/) |
| ai-elements-vue | ^1.4.0 | [ai-elements-vue.com](https://ai-elements-vue.com/) |

### 后端（核心特色）

| 技术 | 版本 | 文档 |
| --- | --- | --- |
| Python | 3.11+ | [python.org](https://www.python.org/) |
| FastAPI | 0.104.1 | [fastapi.tiangolo.com](https://fastapi.tiangolo.com/) |
| LangChain | **1.0.0** | [python.langchain.com](https://python.langchain.com/) |
| LangGraph | 1.0.3 | [langchain-ai.github.io/langgraph](https://langchain-ai.github.io/langgraph/) |
| **SSE 流式协议** | - | [MDN SSE](https://developer.mozilla.org/zh-CN/docs/Web/API/Server-sent_events) |
| **AG-UI 事件规范** | 0.1.18 | [ag-ui.com](https://docs.ag-ui.com/) |
| 阿里云 Qwen TTS | - | [DashScope](https://help.aliyun.com/zh/model-studio/user-guide/tts/) |

> **后端三大特色**：LangChain 1.0 全新 API 体系、SSE 流式实时通信、AG-UI 标准事件协议，三者组合实现了真正的 AI Agent 流式交互体验。

***

## 关键技术说明

### 1\. 前端框架：Vue 3 \+ TypeScript

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

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

***

### 2\. UI 组件：Tailwind CSS \+ shadcn\-vue \+ ai\-elements\-vue

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

```text

<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>
```

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

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

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

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

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

<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>
```

***

### 3\. 后端框架：FastAPI

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

```python
# 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(),
 )
```

***

### 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）只需要改环境变量，不需要动业务代码：

```python
# 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，避免提示词和代码不一致：

```python
# 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 = "\n".join(tool_descriptions)
full_prompt = get_full_prompt(tools_list_text) # 注入到系统提示词
```

#### Agent 创建与工具注册

```python
# 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` 只适合本地简单尝试，真实业务需要用数据库
```python
# 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` 自动恢复，接口保持简洁。

***

### 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 逻辑完全解耦：

```python
# 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`，统一解析事件类型，把文本增量和工具结果分别回调给上层：

```typescript
// 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：

```typescript
// 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（文本打字机 / 工具结果卡片）
```

***

## 整体架构图

```
┌─────────────────────────────────────────┐
│ 前端（Vue 3） │
│ 对话界面 → api/chat.ts → SSE 长连接 │
└───────────────┬─────────────────────────┘
 │ POST /api/chat
┌───────────────▼─────────────────────────┐
│ 后端（FastAPI） │
│ routers/chat.py → agent_service.py │
│ ↓ │
│ LangGraph Agent │
│ ┌──────────┬──────────────┐ │
│ │ LLM 推理 │ 工具调用判断 │ │
│ └──────────┴──────┬───────┘ │
│ ↓ │
│ tools/qwen_tts.py │
│ ┌───────────────────────────┐ │
│ │ voice_design / voice_ │ │
│ │ cloning（阿里云 DashScope）│ │
│ └──────────────┬────────────┘ │
│ ↓ 保存音频文件 │
│ storage/audios/ │
│ ↓ │
│ stream_processor.py │
│ AG-UI 事件编码 → SSE 推流 │
└─────────────────────────────────────────┘
```

***

## 作为新手，我踩过的坑

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` 隔离是必须做的事。

***

## 功能展示

> 以下是项目运行效果截图 👇

* 音频创作对话效果
![image.png](https://s3.cn-north-1.jdcloud-oss.com/shendengbucket1/2026-05-27-16-33BKKpdfeeDjEagZO.png)

* 音色复刻功能
![](https://s3.cn-north-1.jdcloud-oss.com/shendengbucket1/2026-05-27-16-36c0Okvc5e8fNgW5r.png)

***

## 总结

这个项目让我第一次把前端、后端、AI Agent 三块内容独立串联起来。技术选型上每一块都是当下主流的方案，实际跑通整条链路之后对"全栈"有了更具体的感受。

对于同样在学习 AI 应用开发的朋友，有几点心得：

* **框架版本要锁定**：LangChain 这类快速迭代的框架，版本差异带来的问题远比你想象的多
* **先跑通链路，再追求完美**：新手阶段能把用户输入 → AI 处理 → 结果展示这条链跑通，比深入某个技术点更有价值
* **多看官方文档**：相比博客和教程，官方文档更新及时，特别是 LangGraph 这种新框架

后续计划继续迭代，欢迎交流！