在 AI 时代,单人全栈开发的边界正在被重新定义。本文将带你从一个空仓库出发,构建一个由多 Agent 协作驱动的全栈应用——前端用 React 19,后端用 Bun + Elysia,再用一套轻量级 DevOps 流水线完成交付。
过去我们谈“全栈”,指的是一个人能写前端、后端、数据库和部署脚本。今天,当大模型以 Agent 形态介入开发流程后,“全栈”的内涵发生了质变——你不再只是写代码的人,更是 Agent 系统的架构师与协调者。
本文的实战目标很具体:用一个人、一台电脑、一个周末,交付一个可运行的 AI 原生全栈应用。
技术选型上,我们不搞大杂烩,而是精挑一套“少而强”的现代栈:
读完本文,你将收获:
在写任何代码之前,我们先明确系统的边界与交互方式。
┌─────────────────────────────────────────────────────────────┐
│ 前端层 (React 19) │
│ ┌──────────┐ ┌──────────┐ ┌──────────────────────────┐ │
│ │ UI 组件 │←→│ Actions │←→│ useOptimistic / useForm │ │
│ └──────────┘ └──────────┘ └──────────────────────────┘ │
└──────────────────────────┬──────────────────────────────────┘
│ HTTP (REST + SSE)
┌──────────────────────────▼──────────────────────────────────┐
│ API 网关 (Elysia) │
│ ┌────────────────────────────────────────────────────────┐ │
│ │ /api/chat → Agent Orchestrator │ │
│ │ /api/task → 任务管理 │ │
│ │ /api/stream → SSE 流式输出 │ │
│ └────────────────────────────────────────────────────────┘ │
└──────────────────────────┬──────────────────────────────────┘
│
┌──────────────────────────▼──────────────────────────────────┐
│ 多 Agent 协作层 (核心) │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Orchestrator │─→│ Worker A │ │ Worker B │ │
│ │ (调度器) │ │ (代码生成) │ │ (代码审查) │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │ │ │
│ └──────────────┴─────────────────┘ │
│ │ │
│ ┌───────▼───────┐ │
│ │ LLM 网关层 │ │
│ │ (OpenAI/本地) │ │
│ └───────────────┘ │
└─────────────────────────────────────────────────────────────┘用户在前端输入一个 Prompt → React Action 发起请求 → Elysia 路由转发至 Orchestrator Agent → Orchestrator 拆解任务并分发给多个 Worker Agents → 各 Worker 并行/串行执行 → 结果聚合 → 通过 SSE 流式回传前端 → React 利用 useOptimistic 即时更新 UI。
核心设计原则:
为什么选 Bun + Elysia?在 Node.js 生态里,Bun 提供了开箱即用的 TypeScript 支持和惊人的启动速度;Elysia 则是一个“E2E 类型安全”的框架,其 Eden 插件能直接从前端调用后端 API 时获得完整类型提示。
mkdir ai-fullstack && cd ai-fullstack
bun init -y
bun add elysia @elysiajs/cors @elysiajs/stream
bun add -d @types/bun首先,我们定义 Agent 之间的通信契约:
// src/types/agent.ts
export interface AgentMessage {
id: string;
role: 'user' | 'assistant' | 'system' | 'worker';
content: string;
metadata?: Record<string, any>;
}
export interface AgentTask {
id: string;
type: 'code_generation' | 'code_review' | 'summarization' | 'translation';
input: string;
context?: Record<string, any>;
}
export interface AgentResult {
taskId: string;
output: string;
confidence?: number;
tokensUsed?: number;
}为了不绑定单一模型,我们抽象一层 LLM Gateway:
// src/llm/gateway.ts
export interface LLMProvider {
chat(messages: AgentMessage[], options?: any): AsyncIterable<string>;
}
export class OpenAIGateway implements LLMProvider {
constructor(private apiKey: string, private model = 'gpt-4o-mini') {}
async *chat(messages: AgentMessage[]): AsyncIterable<string> {
const response = await fetch('https://api.openai.com/v1/chat/completions', {
method: 'POST',
headers: {
'Authorization': `Bearer ${this.apiKey}`,
'Content-Type': 'application/json',
},
body: JSON.stringify({
model: this.model,
messages: messages.map(m => ({ role: m.role, content: m.content })),
stream: true,
}),
});
const reader = response.body?.getReader();
const decoder = new TextDecoder();
while (reader) {
const { done, value } = await reader.read();
if (done) break;
const chunk = decoder.decode(value);
const lines = chunk.split('\n').filter(line => line.startsWith('data: '));
for (const line of lines) {
const payload = line.replace('data: ', '');
if (payload === '[DONE]') continue;
try {
const parsed = JSON.parse(payload);
const content = parsed.choices?.[0]?.delta?.content;
if (content) yield content;
} catch (e) {
// 忽略解析错误
}
}
}
}
}Orchestrator 是整个系统的“大脑”。它接收用户请求,决定调用哪些 Worker,并聚合结果。
// src/agents/orchestrator.ts
import { AgentMessage, AgentTask, AgentResult } from '../types/agent';
import { LLMProvider } from '../llm/gateway';
export class OrchestratorAgent {
constructor(private llm: LLMProvider) {}
async *execute(userInput: string): AsyncIterable<string> {
// Step 1: 意图识别与任务拆解
const tasks = await this.decomposeTask(userInput);
yield `[系统] 已将任务拆解为 ${tasks.length} 个子任务\n`;
// Step 2: 分发任务给 Workers(并行执行)
const workerPromises = tasks.map(task => this.runWorker(task));
const results = await Promise.all(workerPromises);
// Step 3: 聚合结果
const aggregated = results.map(r => r.output).join('\n');
yield `[系统] 所有子任务已完成,正在聚合...\n`;
// Step 4: 最终润色
const finalResponse = await this.polishResult(userInput, aggregated);
for await (const chunk of finalResponse) {
yield chunk;
}
}
private async decomposeTask(input: string): Promise<AgentTask[]> {
// 这里调用 LLM 进行任务拆解,返回结构化任务列表
// 实际实现中会用 JSON Mode 或 Function Calling
return [
{ id: crypto.randomUUID(), type: 'code_generation', input: '生成 React 组件' },
{ id: crypto.randomUUID(), type: 'code_review', input: '审查生成的代码' },
];
}
private async runWorker(task: AgentTask): Promise<AgentResult> {
// 根据 task.type 路由到不同的 Worker
// 这里简化处理,直接调用 LLM
let output = '';
for await (const chunk of this.llm.chat([
{ id: '1', role: 'system', content: `你是一个${task.type}专家` },
{ id: '2', role: 'user', content: task.input },
])) {
output += chunk;
}
return { taskId: task.id, output };
}
private async *polishResult(input: string, aggregated: string): AsyncIterable<string> {
const messages: AgentMessage[] = [
{ id: '1', role: 'system', content: '请将以下内容润色为面向用户的友好回复' },
{ id: '2', role: 'user', content: `原始需求:${input}\n\n聚合结果:${aggregated}` },
];
for await (const chunk of this.llm.chat(messages)) {
yield chunk;
}
}
}// src/index.ts
import { Elysia, t } from 'elysia';
import { cors } from '@elysiajs/cors';
import { stream } from '@elysiajs/stream';
import { OrchestratorAgent } from './agents/orchestrator';
import { OpenAIGateway } from './llm/gateway';
const app = new Elysia()
.use(cors())
.use(stream())
.decorate('orchestrator', new OrchestratorAgent(new OpenAIGateway(process.env.OPENAI_API_KEY!)))
.post('/api/chat', async ({ body, orchestrator, set }) => {
const { message } = body as { message: string };
set.headers['Content-Type'] = 'text/event-stream';
return stream(async (stream) => {
for await (const chunk of orchestrator.execute(message)) {
stream.send(chunk);
}
stream.send('[DONE]');
});
}, {
body: t.Object({ message: t.String() }),
})
.get('/health', () => ({ status: 'ok' }))
.listen(3000);
console.log(`🦊 Elysia running at http://localhost:3000`);关键点:使用 @elysiajs/stream 插件,我们可以轻松将 Agent 的异步生成器(AsyncGenerator)转换为 SSE 流,前端可以逐字接收。
React 19 带来的最大变化是 Actions 和 useOptimistic 的深度融合,这让 AI 应用的交互体验上了一个台阶——用户提交 Prompt 后,UI 可以立即进入“乐观更新”状态,无需等待服务端响应。
cd frontend
bun create vite . --template react-ts
bun add react@rc react-dom@rc
bun add -D @types/react@npm:types-react@rcReact 19 的 useActionState 和 useFormStatus 让表单处理变得异常简洁:
// src/hooks/useChat.ts
import { useActionState, useOptimistic, useRef } from 'react';
type ChatState = {
messages: Array<{ role: 'user' | 'assistant'; content: string }>;
isStreaming: boolean;
};
export function useChat() {
const [state, setState] = useActionState(
async (prevState: ChatState, formData: FormData) => {
const userMessage = formData.get('message') as string;
if (!userMessage) return prevState;
// 添加用户消息
const newMessages = [...prevState.messages, { role: 'user', content: userMessage }];
setState((prev) => ({ ...prev, messages: newMessages, isStreaming: true }));
// 发起 SSE 请求
const response = await fetch('/api/chat', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ message: userMessage }),
});
const reader = response.body?.getReader();
const decoder = new TextDecoder();
let assistantContent = '';
while (reader) {
const { done, value } = await reader.read();
if (done) break;
const chunk = decoder.decode(value);
const lines = chunk.split('\n').filter(line => line.startsWith('data: '));
for (const line of lines) {
const payload = line.replace('data: ', '');
if (payload === '[DONE]') {
setState((prev) => ({ ...prev, isStreaming: false }));
continue;
}
assistantContent += payload;
// 实时更新消息列表
setState((prev) => ({
...prev,
messages: [
...prev.messages.slice(0, -1),
{ role: 'assistant', content: assistantContent },
],
}));
}
}
return {
messages: [...newMessages, { role: 'assistant', content: assistantContent }],
isStreaming: false,
};
},
{ messages: [], isStreaming: false }
);
return state;
}注意:由于
useActionState目前对异步迭代器的支持有限,上述代码中我们在 Action 内部手动处理 SSE 流,并通过多次setState实现逐字渲染。
在 AI 对话场景中,用户提交消息后,我们希望在服务端响应之前,UI 立即显示“思考中...”状态:
// src/components/ChatBox.tsx
import { useOptimistic, useRef } from 'react';
import { useChat } from '../hooks/useChat';
export function ChatBox() {
const { messages, isStreaming } = useChat();
const [optimisticMessages, addOptimisticMessage] = useOptimistic(
messages,
(state, newMessage: string) => [...state, { role: 'user', content: newMessage, isOptimistic: true }]
);
const formRef = useRef<HTMLFormElement>(null);
const handleSubmit = (formData: FormData) => {
const message = formData.get('message') as string;
if (!message) return;
// 乐观更新:立即在 UI 中显示用户消息
addOptimisticMessage(message);
// 实际提交由 useActionState 处理
formRef.current?.submit();
};
return (
<div>
<div className="messages">
{optimisticMessages.map((msg, idx) => (
<div key={idx} className={msg.role}>
{msg.content}
{msg.isOptimistic && <span className="spinner">⏳</span>}
</div>
))}
{isStreaming && <div className="assistant typing">正在思考...</div>}
</div>
<form ref={formRef} action={handleSubmit}>
<input name="message" placeholder="输入你的需求..." />
<button type="submit" disabled={isStreaming}>发送</button>
</form>
</div>
);
}Elysia 的 Eden 插件可以让前端直接获得后端 API 的类型定义,实现真正的端到端类型安全:
// frontend/src/api/client.ts
import { treaty } from '@elysiajs/eden';
import type { App } from '../../backend/src/index'; // 引用后端类型
export const client = treaty<App>('http://localhost:3000');
// 使用时
const { data, error } = await client.api.chat.post({ message: 'Hello' });很多人把“多 Agent”简单理解为多次调用 LLM,但这远远不够。真正有生产力的多 Agent 系统需要具备:
我们不需要完整引入 LangChain,而是借鉴其 Graph 思想,实现一个轻量级的 Agent 状态机:
// src/agents/graph.ts
export type AgentNode = {
id: string;
execute: (input: any) => Promise<any>;
dependencies?: string[];
};
export class AgentGraph {
private nodes: Map<string, AgentNode> = new Map();
addNode(node: AgentNode) {
this.nodes.set(node.id, node);
}
async execute(entryNodeId: string, initialInput: any): Promise<Map<string, any>> {
const results = new Map<string, any>();
const visited = new Set<string>();
const dfs = async (nodeId: string) => {
if (visited.has(nodeId)) return;
visited.add(nodeId);
const node = this.nodes.get(nodeId);
if (!node) throw new Error(`Node ${nodeId} not found`);
// 先执行依赖节点
if (node.dependencies) {
for (const depId of node.dependencies) {
await dfs(depId);
}
}
// 收集依赖的输出作为输入
const input = node.dependencies
? Object.fromEntries(node.dependencies.map(depId => [depId, results.get(depId)]))
: initialInput;
const output = await node.execute(input);
results.set(nodeId, output);
};
await dfs(entryNodeId);
return results;
}
}// src/agents/workers.ts
export const codeGeneratorWorker: AgentNode = {
id: 'code_generator',
dependencies: [],
async execute(input: { requirement: string }) {
// 调用 LLM 生成代码
return { code: '...', language: 'typescript' };
},
};
export const codeReviewerWorker: AgentNode = {
id: 'code_reviewer',
dependencies: ['code_generator'],
async execute(input: { code_generator: { code: string } }) {
// 审查生成的代码
return { feedback: '...', score: 85 };
},
};
export const testGeneratorWorker: AgentNode = {
id: 'test_generator',
dependencies: ['code_generator'],
async execute(input: { code_generator: { code: string } }) {
// 生成单元测试
return { tests: '...' };
},
};// src/agents/orchestrator.ts (增强版)
export class OrchestratorAgent {
private graph = new AgentGraph();
constructor(private llm: LLMProvider) {
this.graph.addNode(codeGeneratorWorker);
this.graph.addNode(codeReviewerWorker);
this.graph.addNode(testGeneratorWorker);
}
async *execute(userInput: string): AsyncIterable<string> {
// 用 LLM 解析用户意图,决定执行哪个 Graph 入口
const entryNode = await this.selectEntryNode(userInput);
const results = await this.graph.execute(entryNode, { requirement: userInput });
// 格式化输出
for (const [nodeId, result] of results) {
yield `[${nodeId}] ${JSON.stringify(result)}\n`;
}
}
}单人全栈最容易踩的坑就是“过度工程化 DevOps”。我的策略是:用最少工具覆盖核心需求。
# Dockerfile
FROM oven/bun:latest AS builder
WORKDIR /app
COPY package.json bun.lockb ./
RUN bun install --frozen-lockfile
COPY . .
RUN bun run build
FROM oven/bun:latest AS runner
WORKDIR /app
COPY --from=builder /app/dist ./dist
COPY --from=builder /app/node_modules ./node_modules
COPY package.json ./
ENV NODE_ENV=production
EXPOSE 3000
CMD ["bun", "run", "dist/index.js"]# docker-compose.yml
version: '3.8'
services:
backend:
build: .
ports:
- "3000:3000"
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
restart: unless-stopped
frontend:
build: ./frontend
ports:
- "5173:5173"
depends_on:
- backend
environment:
- VITE_API_URL=http://backend:3000# .github/workflows/deploy.yml
name: Deploy to VPS
on:
push:
branches: [main]
jobs:
deploy:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Setup Bun
uses: oven-sh/setup-bun@v1
- name: Install dependencies
run: bun install
- name: Run tests
run: bun test
- name: Build
run: bun run build
- name: Deploy to VPS
uses: appleboy/scp-action@v0.1.7
with:
host: ${{ secrets.VPS_HOST }}
username: ${{ secrets.VPS_USER }}
key: ${{ secrets.VPS_SSH_KEY }}
source: "dist/,package.json,bun.lockb,Dockerfile,docker-compose.yml"
target: "/app/ai-fullstack"
- name: Restart service
uses: appleboy/ssh-action@v1.0.3
with:
host: ${{ secrets.VPS_HOST }}
username: ${{ secrets.VPS_USER }}
key: ${{ secrets.VPS_SSH_KEY }}
script: |
cd /app/ai-fullstack
docker-compose down
docker-compose up -d --build单人项目不需要 Prometheus + Grafana 的重型组合。用 bun:serve 的内置日志 + 简单的健康检查即可:
// 在 Elysia 中添加健康检查和内存监控
app.get('/metrics', () => ({
memory: process.memoryUsage(),
uptime: process.uptime(),
}));配合 cron 定时任务做简单告警(通过 Telegram Bot 推送)。
当多个 Worker 同时调用 LLM API 时,容易触发速率限制(Rate Limit)。解决方案:引入一个简单的 Semaphore(信号量):
// src/utils/semaphore.ts
export class Semaphore {
private permits: number;
private queue: Array<() => void> = [];
constructor(permits: number) {
this.permits = permits;
}
async acquire(): Promise<void> {
if (this.permits > 0) {
this.permits--;
return;
}
return new Promise(resolve => this.queue.push(resolve));
}
release() {
this.permits++;
const next = this.queue.shift();
if (next) next();
}
}
// 使用:全局限制并发为 5
const llmSemaphore = new Semaphore(5);逐字渲染虽然体验好,但如果 LLM 输出过快(比如 100 tokens/s),频繁的 setState 会导致掉帧。解决方案:使用 requestAnimationFrame 节流:
let pendingUpdate = false;
const throttledSetState = (newContent: string) => {
if (!pendingUpdate) {
pendingUpdate = true;
requestAnimationFrame(() => {
setState(prev => ({ ...prev, content: newContent }));
pendingUpdate = false;
});
}
};相同或相似的 Prompt 不应该重复调用 LLM。使用 node-cache 做简单的内容缓存:
import NodeCache from 'node-cache';
const cache = new NodeCache({ stdTTL: 3600 });
// 在 LLM Gateway 中
async *chat(messages: AgentMessage[]): AsyncIterable<string> {
const cacheKey = JSON.stringify(messages);
const cached = cache.get<string>(cacheKey);
if (cached) {
yield cached;
return;
}
// ... 调用 API,并将完整结果缓存
}AI 应用的测试不同于传统应用,我们主要关注:
// test/agents/orchestrator.test.ts
import { describe, expect, test, mock } from 'bun:test';
import { OrchestratorAgent } from '../../src/agents/orchestrator';
describe('OrchestratorAgent', () => {
test('应该正确拆解任务', async () => {
const mockLLM = {
chat: mock(async function* () {
yield '拆解为 2 个任务';
}),
};
const orchestrator = new OrchestratorAgent(mockLLM as any);
let result = '';
for await (const chunk of orchestrator.execute('生成一个登录页面')) {
result += chunk;
}
expect(result).toContain('2 个子任务');
});
});// test/e2e/chat-flow.test.ts
import { describe, expect, test } from 'bun:test';
describe('E2E Chat Flow', () => {
test('完整对话流程', async () => {
const response = await fetch('http://localhost:3000/api/chat', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ message: 'Hello' }),
});
expect(response.status).toBe(200);
expect(response.headers.get('content-type')).toContain('text/event-stream');
});
});在 AI 时代,“一人全栈”不再意味着你要成为所有领域的专家,而是你要懂得:
本文构建的系统只是一个起点。你可以在它的基础上继续扩展:
真正的全栈,不是“什么都会”,而是“知道如何用工具和 AI 把事做成”。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。