第四章：内容生成工作流

用户：分析 Vue3 和 React，生成报告
Agent 自己决定：
  → 调 web_search("Vue3")
  → 调 web_search("React")
  → 调 write_report(...)
  完成

周报生成工作流，步骤永远是这样：
  步骤1：提炼工作亮点（固定）
  步骤2：识别风险阻塞（固定）
  步骤3：等人工确认（固定）
  步骤4：生成周报（固定）

const State = Annotation.Root({
  // 用户输入
  points: Annotation({ reducer: (_, n) => n, default: () => '' }),
  // 第一步的输出
  highlights: Annotation({ reducer: (_, n) => n, default: () => '' }),
  // 第二步的输出
  risks: Annotation({ reducer: (_, n) => n, default: () => '' }),
  // 最终结果
  report: Annotation({ reducer: (_, n) => n, default: () => '' }),
})

async function extractHighlights(state) {
  // 读入状态（用用户输入的工作要点）
  const result = await callLLM(state.points)

  // 只返回需要更新的字段
  // 注意：不需要返回完整的 state，只返回有变化的字段
  return { highlights: result }
}

// 普通边：A 完成后，一定执行 B
.addEdge('node_a', 'node_b')

// 条件边：A 完成后，根据状态决定走 B 还是 C
.addConditionalEdges('node_a', routeFunction, {
  'go_b': 'node_b',
  'go_c': 'node_c',
})

import { StateGraph, END, START, Annotation } from '@langchain/langgraph'
import { MemorySaver } from '@langchain/langgraph'

// 第一步：定义状态
const State = Annotation.Root({
  points:    Annotation({ reducer: (_, n) => n, default: () => '' }),
  highlights: Annotation({ reducer: (_, n) => n, default: () => '' }),
  risks:      Annotation({ reducer: (_, n) => n, default: () => '' }),
  report:     Annotation({ reducer: (_, n) => n, default: () => '' }),
  humanFeedback: Annotation({ reducer: (_, n) => n, default: () => '' }),
})

// 第二步：定义节点函数
async function extractHighlights(state) {
  const result = await callLLM(
    '从工作要点中提炼3-5条亮点',
    state.points
  )
  return { highlights: result }
}

async function identifyRisks(state) {
  const result = await callLLM(
    '识别风险和阻塞项',
    state.points
  )
  return { risks: result }
}

// 这个函数什么都不做，是一个"占位符"
// 实际暂停行为由 interruptBefore 配置控制
async function humanReview(state) {
  return {}
}

async function generateReport(state) {
  const result = await callLLM(
    `生成周报，用户反馈：${state.humanFeedback}`,
    `亮点：${state.highlights}\n风险：${state.risks}`
  )
  return { report: result }
}

// 第三步：构建图
const checkpointer = new MemorySaver()  // 存档点，用于暂停/恢复

const graph = new StateGraph(State)
  .addNode('extract_highlights', extractHighlights)
  .addNode('identify_risks',     identifyRisks)
  .addNode('human_review',       humanReview)
  .addNode('generate_report',    generateReport)
  .addEdge(START,                'extract_highlights')
  .addEdge('extract_highlights', 'identify_risks')
  .addEdge('identify_risks',     'human_review')
  .addEdge('human_review',       'generate_report')
  .addEdge('generate_report',     END)
  .compile({
    checkpointer,
    interruptBefore: ['human_review'],   // ← 关键！在这个节点前暂停
  })

graph.compile({
  checkpointer,                         // 必须有存档点
  interruptBefore: ['human_review'],    // 在执行 human_review 节点之前暂停
})

// 启动工作流（会自动在 human_review 前暂停）
for await (const event of graph.streamEvents(input, { ...config, version: 'v2' })) {
  // 处理事件...
}
// 到这里，图已经暂停了，不是真的完成了

// 检查当前状态
const state = await graph.getState(config)
console.log(state.next)    // ['human_review'] ← 说明还没执行 human_review，暂停了
console.log(state.values)  // 包含已执行节点的输出（highlights、risks 等）

// 1. 注入人工反馈（可选）
if (feedback) {
  await graph.updateState(config, { humanFeedback: feedback })
}

// 2. 传 null 给 streamEvents，从暂停处继续执行
for await (const event of graph.streamEvents(null, { ...config, version: 'v2' })) {
  // 继续处理后续节点的事件...
}

用户填写输入内容
  ↓
[前端] POST /api/workflow/start/stream
  ↓
[服务端] 执行 extractHighlights → identifyRisks
  ↓
[服务端] 到 human_review 前自动暂停
  ↓
SSE 推送 → event: paused，附带 threadId + 中间产物
  ↓
[前端] 展示中间产物，显示 HumanReviewPanel
  ↓
用户查看亮点/风险，可选填写"这里写得不对，帮我换一下"
  ↓
[前端] POST /api/workflow/resume/stream（带 threadId 和 feedback）
  ↓
[服务端] 注入 humanFeedback 到状态，继续执行 generateReport
  ↓
SSE 流式推送最终周报内容
  ↓
[前端] event: completed，展示结果

// 用 Map 存活跃的工作流实例
const activeWorkflows = new Map()  // threadId → { graph, meta, config }

// 启动时存入
const graph = builder()
activeWorkflows.set(threadId, { graph, meta, config })

// 恢复时取出
const wf = activeWorkflows.get(threadId)
if (!wf) {
  return res.status(404).json({ error: '工作流不存在或已过期，请重新启动' })
}

// 完成后清理
activeWorkflows.delete(threadId)

for await (const event of graph.streamEvents(input, { ...config, version: 'v2' })) {
  const { event: eventType, name } = event

  // 节点开始执行（过滤掉 LangGraph 内部节点）
  if (eventType === 'on_chain_start') {
    const nodeInMeta = meta.nodes.find(n => n.id === name)
    if (nodeInMeta) {
      send('node_start', { nodeId: name, label: nodeInMeta.label })
    }
  }

  // 节点执行完毕
  if (eventType === 'on_chain_end') {
    const nodeInMeta = meta.nodes.find(n => n.id === name)
    if (nodeInMeta) {
      // 提取节点输出的预览文本（给前端在节点卡片上显示）
      const output = event.data?.output
      const preview = getPreview(output)
      send('node_done', { nodeId: name, preview })
    }
  }
}

// 遍历完 streamEvents 之后检查状态
const state = await graph.getState(config)

if (state.next?.length > 0) {
  // state.next 不为空 → 还有待执行的节点 → 说明是暂停的
  // state.next[0] 就是下一个要执行的节点名
  send('paused', {
    threadId,
    nextNode: state.next[0],
    intermediates: getIntermediates(state.values, workflowId),
  })
} else {
  // state.next 为空 → 所有节点都执行完了 → 真正完成
  send('completed', { threadId, result: state.values[meta.resultKey] })
}

idle（初始）
  ↓ 用户点击开始
running（执行中）
  ↓ 到达 human_review
paused（等待人工审核）
  ↓ 用户点击继续
running（继续执行最后节点）
  ↓ 完成
completed（展示结果）

const running       = ref(false)  // 是否在执行
const paused        = ref(false)  // 是否暂停在审核
const result        = ref('')     // 最终结果
const streamBuffer  = ref('')     // 流式输出的缓冲

// nodeStates 存每个节点的状态
const nodeStates = reactive({})
// { 'extract_highlights': 'done', 'identify_risks': 'running', ... }
// 可能的值：'idle' | 'running' | 'done' | 'waiting'

// nodeOutputs 存每个节点的输出预览
const nodeOutputs = reactive({})
// { 'extract_highlights': '• 完成了需求评审...\n• 发布了新版本...' }

<div
  class="node-card"
  :class="nodeState(node.id)"  <!-- 'idle' | 'running' | 'done' | 'waiting' -->
>
  <!-- 状态圆圈 -->
  <div class="node-status-circle" :class="nodeState(node.id)">
    <span v-if="nodeState(node.id) === 'done'"    >✓</span>
    <span v-else-if="nodeState(node.id) === 'running'" class="spinner-sm" />
    <span v-else-if="nodeState(node.id) === 'waiting'" >⏸</span>
    <span v-else>{{ idx + 1 }}</span>
  </div>
</div>

<div class="connector-line" :class="{ lit: nodeState(node.id) === 'done' }" />

/* 默认灰色 */
.connector-line { background: var(--color-border); }
/* 上一个节点完成后，变绿色 */
.connector-line.lit { background: var(--color-success); }

await fetchStream('/api/workflow/start/stream', payload, {
  onEvent: (event, data) => {
    if (event === 'node_start') {
      // 把这个节点标记为执行中
      nodeStates[data.nodeId] = 'running'
    }

    if (event === 'node_done') {
      // 节点执行完，更新为完成
      nodeStates[data.nodeId] = 'done'
      // 如果有预览内容，一起存下来
      if (data.preview) nodeOutputs[data.nodeId] = data.preview
    }

    if (event === 'paused') {
      // 工作流暂停在审核节点
      paused.value = true
      running.value = false
      nodeStates['human_review'] = 'waiting'
      // 存下来供审核面板展示
      intermediates.value = data.intermediates || []
    }

    if (event === 'completed') {
      result.value = data.result
      running.value = false
    }
  }
})

<div class="review-panel">
  <!-- 展示中间产物 -->
  <div v-for="item in intermediates" :key="item.key" class="intermediate-item">
    <div class="item-label">{{ item.label }}</div>
    <div class="item-value">{{ item.value }}</div>
  </div>

  <!-- 修改意见 -->
  <textarea v-model="feedback" placeholder="如有问题，在此填写修改意见..." />

  <!-- 操作 -->
  <button @click="emit('approve', feedback)">
    {{ feedback.trim() ? '采纳意见并继续' : '确认并继续' }}
  </button>
</div>

async function resumeWithFeedback(feedback) {
  await wfStore.resumeWorkflow(feedback)
}

async function resumeWorkflow(feedback) {
  // 1. 请求恢复接口
  await fetchStream('/api/workflow/resume/stream', { threadId, feedback }, {
    onToken: (token) => {
      // 最后一个节点流式输出
      streamBuffer.value += token
    },
    onEvent: (event, data) => {
      if (event === 'node_start') nodeStates[data.nodeId] = 'running'
      if (event === 'node_done')  nodeStates[data.nodeId] = 'done'
      if (event === 'completed')  result.value = data.result
    }
  })
}

// 第一步：专注提炼亮点
async function extractHighlights(state) {
  return chatLLM(`
    从工作要点中提炼亮点。
    输出3-5条，每条一行，用"• "开头，不超过30字。
    不要发散，只提炼原文中有的信息。
  `, state.points)
}

// 第二步：专注风险
async function identifyRisks(state) {
  return chatLLM(`
    从工作内容中识别风险和阻塞项。
    如果没有明显风险，输出"本周无明显风险项"。
    只看负面信息，不要混入成果。
  `, state.points)
}

提取参会人/议题（最简单，先做）
  ↓
提取结论（需要理解全文）
  ↓
整理 Action Items（最复杂，最后做）
  ↓
人工审核（确认结论和 action 是否正确）
  ↓
生成正式纪要

分析语气/目的（给后续步骤提供方向）
  ↓
检查问题（找出具体的措辞问题）
  ↓
人工审核（确认问题分析是否准确，避免把好的改坏了）
  ↓
生成润色版本（基于分析结果优化）

system: 你是周报助手
user: 帮我生成周报，以下是工作要点：[内容]
      要求：先提炼亮点，再识别风险，最后生成完整周报

问题1：模型可能不按你期望的顺序走，直接输出最终结果跳过中间分析
问题2：没有办法在中途让用户干预
问题3：中间产物（亮点、风险）没法单独展示给用户
问题4：某一步出错了，需要整个重来

✓ 每一步的 prompt 更精简聚焦，效果更好
✓ 中间产物可以单独展示，用户能看到"AI 提炼的亮点是什么"
✓ 人工干预点明确（就在 human_review 节点）
✓ 某一步出错，可以从那步重新来，不用整个重跑
✓ 可以给每一步设置不同的 temperature（分析步骤用 0，生成步骤用 0.7）

import { PostgresSaver } from '@langchain/langgraph-checkpoint-postgres'

const checkpointer = PostgresSaver.fromConnString(process.env.DATABASE_URL)
await checkpointer.setup()   // 初始化表结构

const output = event.data?.output
const firstVal = Object.values(output)[0]
if (typeof firstVal === 'string') preview = firstVal.slice(0, 80)

async function extractHighlights(state) {
  try {
    const result = await callLLM(...)
    return { highlights: result }
  } catch (e) {
    // 失败时给一个默认值，让工作流继续
    return { highlights: '（亮点提取失败，请手动填写）' }
  }
}

const threadId = `wf_${Date.now()}_${Math.random().toString(36).slice(2, 6)}`
const config   = { configurable: { thread_id: threadId } }