AI 智能体与代理任务

ProductReady 包含一个生产就绪的任务管理系统，专为 AI 智能体设计。非常适合：

🤖 AI 自动化工作流
📋 后台作业处理
🎯 用户分配的任务
🔄 异步操作跟踪

生产就绪：开箱即用包含数据库模式、tRPC API 和仪表板 UI！

包含内容

✅ 数据库模式 - 具有优先级、状态和分配的任务
✅ tRPC API - 完整的 CRUD 操作
✅ 仪表板 UI - 任务列表和创建表单
✅ 演示模式 - 无需数据库设置即可试用
✅ 类型安全 - 端到端 TypeScript

快速开始

查看现有任务

启动开发服务器：pnpm dev
登录仪表板：http://localhost:3000/dashboard
您将看到任务仪表板！

或尝试演示模式（无需登录）：http://localhost:3000/dashboard?demo=1

数据库模式

任务存储在 PostgreSQL 中：

// src/db/schema/agentTasks.ts
export const agentTasks = pgTable('agentTasks', {
  id: uuid('id').defaultRandom().primaryKey(),
  title: text('title').notNull(),
  description: text('description'),
  status: text('status').notNull().default('pending'),
  priority: text('priority').notNull().default('medium'),
  assignedTo: text('assigned_to'),
  createdAt: timestamp('created_at').defaultNow().notNull(),
  updatedAt: timestamp('updated_at').defaultNow().notNull(),
  completedAt: timestamp('completed_at'),
});

字段说明：

title - 任务名称（必填）
description - 任务详情（可选）
status - pending（待处理）、in_progress（进行中）、completed（已完成）、failed（失败）
priority - low（低）、medium（中）、high（高）、urgent（紧急）
assignedTo - 用户 ID 或智能体名称
createdAt / updatedAt - 时间戳
completedAt - 任务完成时间

使用 API

创建任务

'use client';

import { trpc } from '~/lib/trpc/client';

export function CreateTaskButton() {
  const createTask = trpc.agentTasks.create.useMutation();
  
  async function handleCreate() {
    await createTask.mutateAsync({
      title: '分析客户反馈',
      description: '使用 AI 对 1000 条支持工单进行分类',
      priority: 'high',
      assignedTo: 'ai-agent-1',
    });
    
    alert('任务已创建！');
  }
  
  return <button onClick={handleCreate}>创建 AI 任务</button>;
}

列出所有任务

export function TasksList() {
  const { data: agentTasks, isLoading } = trpc.agentTasks.list.useQuery();
  
  if (isLoading) return <div>加载任务中...</div>;
  
  return (
    <div>
      {agentTasks?.map((task) => (
        <div key={task.id}>
          <h3>{task.title}</h3>
          <p>状态：{task.status}</p>
          <p>优先级：{task.priority}</p>
        </div>
      ))}
    </div>
  );
}

更新任务状态

export function TaskStatusButton({ taskId }: { taskId: string }) {
  const updateTask = trpc.agentTasks.update.useMutation();
  
  async function markComplete() {
    await updateTask.mutateAsync({
      id: taskId,
      status: 'completed',
      completedAt: new Date(),
    });
  }
  
  return <button onClick={markComplete}>标记为完成</button>;
}

构建 AI 智能体

示例：邮件摘要智能体

让我们构建一个可以总结邮件的智能体！

创建智能体函数

// src/lib/agents/email-summarizer.ts
import { OpenAI } from 'openai';

const openai = new OpenAI({
  apiKey: process.env.OPENAI_API_KEY,
});

export async function summarizeEmail(emailContent: string) {
  const response = await openai.chat.completions.create({
    model: 'gpt-4',
    messages: [
      {
        role: 'system',
        content: '你是一个邮件摘要助手。请提供简洁的摘要。',
      },
      {
        role: 'user',
        content: `总结这封邮件：\n\n${emailContent}`,
      },
    ],
  });
  
  return response.choices[0].message.content;
}

创建任务处理器

// src/lib/agents/task-processor.ts
import { db } from '~/db';
import { agentTasks } from '~/db/schema/agentTasks';
import { eq } from 'drizzle-orm';
import { summarizeEmail } from './email-summarizer';

export async function processEmailTask(taskId: string) {
  // 更新状态为进行中
  await db
    .update(agentTasks)
    .set({ status: 'in_progress', updatedAt: new Date() })
    .where(eq(agentTasks.id, taskId));
  
  try {
    // 获取任务详情
    const [task] = await db
      .select()
      .from(agentTasks)
      .where(eq(agentTasks.id, taskId))
      .limit(1);
    
    // 运行 AI 智能体
    const summary = await summarizeEmail(task.description || '');
    
    // 标记为完成并保存结果
    await db
      .update(agentTasks)
      .set({
        status: 'completed',
        completedAt: new Date(),
        description: `原文：\n${task.description}\n\n摘要：\n${summary}`,
      })
      .where(eq(agentTasks.id, taskId));
    
    return { success: true, summary };
  } catch (error) {
    // 标记为失败
    await db
      .update(agentTasks)
      .set({
        status: 'failed',
        description: `错误：${error.message}`,
      })
      .where(eq(agentTasks.id, taskId));
    
    throw error;
  }
}

创建 tRPC 端点触发智能体

// src/server/routers/agentTasks.ts
export const tasksRouter = createTRPCRouter({
  // ... 现有端点 ...
  
  processEmail: protectedProcedure
    .input(z.object({ taskId: z.string().uuid() }))
    .mutation(async ({ ctx, input }) => {
      // 将任务加入处理队列
      await processEmailTask(input.taskId);
      
      return { success: true };
    }),
});

在前端使用

'use client';

export function ProcessEmailButton({ taskId }: { taskId: string }) {
  const processEmail = trpc.agentTasks.processEmail.useMutation();
  
  async function handleProcess() {
    await processEmail.mutateAsync({ taskId });
    alert('邮件处理已启动！');
  }
  
  return <button onClick={handleProcess}>使用 AI 总结</button>;
}

完成！ 您现在拥有一个异步处理任务的 AI 智能体。

后台处理

对于长时间运行的任务，使用作业队列：

选项 1：简单轮询

// 每 10 秒轮询一次待处理的任务
setInterval(async () => {
  const pendingTasks = await db
    .select()
    .from(agentTasks)
    .where(eq(agentTasks.status, 'pending'))
    .limit(10);
  
  for (const task of pendingTasks) {
    await processTask(task.id);
  }
}, 10000);

选项 2：Bull 队列（生产环境）

pnpm add bull @bull-board/api @bull-board/hono

// src/lib/queue.ts
import Bull from 'bull';

export const taskQueue = new Bull('agentTasks', {
  redis: process.env.REDIS_URL,
});

taskQueue.process(async (job) => {
  const { taskId } = job.data;
  await processTask(taskId);
});

// 将作业添加到队列
export async function queueTask(taskId: string) {
  await taskQueue.add({ taskId });
}

任务仪表板

ProductReady 在 /dashboard 包含仪表板 UI。自定义它：

// src/components/dashboard/agentTasks-table.tsx
'use client';

import { trpc } from '~/lib/trpc/client';

export function TasksTable() {
  const { data: agentTasks } = trpc.agentTasks.list.useQuery();
  const deleteTask = trpc.agentTasks.delete.useMutation();
  
  return (
    <table>
      <thead>
        <tr>
          <th>标题</th>
          <th>优先级</th>
          <th>状态</th>
          <th>创建时间</th>
          <th>操作</th>
        </tr>
      </thead>
      <tbody>
        {agentTasks?.map((task) => (
          <tr key={task.id}>
            <td>{task.title}</td>
            <td>
              <PriorityBadge priority={task.priority} />
            </td>
            <td>
              <StatusBadge status={task.status} />
            </td>
            <td>{formatDate(task.createdAt)}</td>
            <td>
              <button onClick={() => deleteTask.mutate({ id: task.id })}>
                删除
              </button>
            </td>
          </tr>
        ))}
      </tbody>
    </table>
  );
}

最佳实践

✅ 推荐做法

跟踪任务状态 - 使用 pending/in_progress/completed/failed
添加时间戳 - createdAt、updatedAt、completedAt
分配所有权 - 知道哪个用户/智能体拥有每个任务
优雅处理失败 - 记录错误，适当时重试
清理旧任务 - 归档或删除已完成的任务

❌ 避免做法

不要阻塞 UI - 对长任务使用后台处理
不要丢失任务数据 - 始终先持久化到数据库
不要跳过错误处理 - 捕获异常并更新任务状态
不要无限期处理 - 设置最大重试次数和超时

使用 Resume Streams 实现后台智能体执行

ProductReady 通过 AI SDK 支持 Resume Streams（恢复流），实现真正的后台智能体任务执行。这意味着即使用户断开连接或关闭浏览器，智能体任务也能继续运行。

突破性能力！ Resume Streams 使智能体能够在后台工作，完成长时间运行的任务，而无需用户保持连接。

工作原理

AI SDK 的恢复流功能允许聊天会话暂停和恢复：

会话持久化：每个聊天会话获得唯一 ID
流缓存：进行中的流在服务器上缓存
自动恢复：如果断开连接，流会自动从中断处恢复
后台执行：即使用户离开，智能体任务也会继续运行

实现方式

ProductReady 的 useAgentChat hook 默认包含恢复支持：

import { useAgentChat } from '~/hooks/use-agent-chat';

export function AgentChatComponent() {
  const { 
    messages, 
    input, 
    handleInputChange, 
    handleSubmit,
    isLoading,
    reload,        // 断开连接后恢复流
    sessionId,     // 此聊天的唯一会话 ID
  } = useAgentChat({
    taskId: 'task-123',
    enableResume: true,  // 启用恢复支持（默认：true）
  });
  
  return (
    <div>
      {/* 聊天 UI */}
      <div>
        {messages.map(msg => (
          <div key={msg.id}>{msg.content}</div>
        ))}
      </div>
      
      {/* 恢复按钮（断开连接后出现） */}
      {!isLoading && messages.length > 0 && (
        <button onClick={() => reload()}>
          恢复流
        </button>
      )}
      
      {/* 输入 */}
      <input value={input} onChange={handleInputChange} />
      <button onClick={handleSubmit}>发送</button>
    </div>
  );
}

主要特性

1. 自动会话管理

const chat = useAgentChat({
  enableResume: true,     // 启用恢复（默认）
  sessionId: 'custom-id', // 可选：提供自己的 ID
});

// 如果未提供，会话 ID 会自动生成
console.log(chat.sessionId); // 例如："550e8400-e29b-41d4-a716-446655440000"

2. 流协议配置

聊天路由自动启用数据流协议：

// 服务器设置兼容恢复的头部
response.headers.set("X-Stream-Protocol", "data");
response.headers.set("X-Session-Id", chatId);

3. 后台任务关联

流自动链接到数据库任务：

const chat = useAgentChat({
  taskId: '123',        // 与现有任务关联
  enableResume: true,
});

// 消息和产物保存到数据库
// 任务在后台继续运行
// 用户可以关闭浏览器并稍后查看结果

使用场景

长时间运行的分析

// 用户请求数据分析
const chat = useAgentChat({
  taskId: analysisTaskId,
  enableResume: true,
});

// 用户发送："分析上个月的销售数据"
// 智能体开始处理（可能需要 5-10 分钟）
// 用户可以关闭页面
// 智能体在后台继续工作
// 用户稍后回来查看结果

多步骤工作流

// 包含多个工具调用的复杂工作流
const chat = useAgentChat({
  taskId: workflowTaskId,
  enableResume: true,
});

// 智能体进行多个 API 调用
// 生成报告
// 创建产物
// 完成时全部保存到数据库

离线弹性

// 用户连接不稳定
const chat = useAgentChat({
  enableResume: true,
});

// 流可以在重新连接后自动恢复
// 不会丢失消息
// 任务从最后检查点继续

服务器端缓存

聊天路由维护会话缓存以支持恢复：

// 内存缓存（开发环境）
// 生产环境请使用 Redis 或类似的分布式缓存
const streamCache = new Map<string, { 
  taskId: number; 
  messages: any[] 
}>();

// 任务完成时自动清理缓存

生产部署：对于多实例部署，请将内存 Map 替换为 Redis 或分布式缓存。

AI 智能体的优势

真正的后台执行：智能体无需用户在场即可工作
可靠性：网络问题不会中断长时间运行的任务
更好的用户体验：用户无需保持浏览器打开
可扩展性：服务器处理多个并发智能体任务
持久性：所有消息和产物保存到数据库

架构流程

用户发起聊天
  ↓
生成会话 ID
  ↓
发送带会话 ID 的消息
  ↓
服务器开始流式响应
  ↓
缓存会话（taskId + 消息）
  ↓
流开始
  ↓
[用户断开连接] ← 可随时发生
  ↓
智能体在后台继续
  ↓
结果保存到数据库
  ↓
用户稍后返回
  ↓
使用相同会话 ID 调用 reload()
  ↓
流从最后位置恢复
  ↓
用户看到完整结果

API 参考

useAgentChat 选项

选项	类型	默认值	描述
`taskId`	`string`	-	将聊天与现有任务关联
`enableResume`	`boolean`	`true`	启用流恢复
`sessionId`	`string`	自动生成	自定义会话 ID
`initialMessages`	`Message[]`	`[]`	起始消息
`onFinish`	`(msg: string) => void`	-	流完成时的回调
`onError`	`(error: Error) => void`	-	错误处理器

返回值

属性	类型	描述
`messages`	`Message[]`	所有聊天消息
`isLoading`	`boolean`	流是否活动
`reload`	`() => void`	断开连接后恢复流
`regenerate`	`() => void`	重新生成最后响应
`sessionId`	`string`	当前会话 ID
`stop`	`() => void`	停止流

下一步

tRPC API 指南 - 构建更多 API 端点
数据库指南 - 高级模式模式
后台作业 - 生产作业队列
测试 - 测试您的智能体

资源

任务路由: src/server/routers/agentTasks.ts
任务模式: src/db/schema/agentTasks.ts
仪表板: src/app/(dashboard)/dashboard/page.tsx

AI 智能体与代理任务

AI 智能体与代理任务

包含内容

快速开始

查看现有任务

数据库模式

使用 API

创建任务

列出所有任务

更新任务状态

构建 AI 智能体

示例：邮件摘要智能体

创建智能体函数

创建任务处理器

创建 tRPC 端点触发智能体

在前端使用

后台处理

选项 1：简单轮询

选项 2：Bull 队列（生产环境）

任务仪表板

最佳实践

✅ 推荐做法

❌ 避免做法

使用 Resume Streams 实现后台智能体执行

工作原理

实现方式

主要特性

使用场景

服务器端缓存

AI 智能体的优势

架构流程

API 参考

相关文档

下一步

资源

On this page