## Detailed Usage Example (AI Flow - RAG + Function Call) ``` typescript import { WorkflowDefinition, createWorkflowContext, WorkflowExecutor, NodeRegistry, InMemoryEventManager, LocalScheduler, ConsoleLogger, StepType, // 引入 FlowLab 核心组件 // 假设 Core 包提供了一些基础节点: LogNode, // 用于在控制台打印日志的节点 HttpRequestNode, // 用于发起 HTTP 请求的节点 } from '@flowlab/core'; // 引入 @flowlab/ai 包的相关组件 import { aiProviderRegistry, // 全局 AI Provider 注册表实例 loadAIConfigFromFile, // (可选) 从文件加载 AI 配置的辅助函数 LLMCompletionNode, // 用于文本生成的 AI 节点 EmbeddingNode, // 用于生成文本向量嵌入的 AI 节点 FunctionCallingNode, // 用于让 AI 决定是否调用以及如何调用外部函数的节点 aiConditionHelper, // 用于在条件步骤中使用 AI 判断的辅助对象 // 假设存在一个用于向量搜索的节点 (可能在 @flowlab/ai 或其他扩展包中) VectorSearchNode, // 输入: 查询向量, TopK; 输出: 相关文档列表[] } from '@flowlab/ai'; // 假设这些都从 @flowlab/ai 导出 // --- 第 1 步: 初始化与配置 --- // 1.1 加载 AI Provider 的配置 (例如，从环境变量或配置文件) // loadAIConfigFromFile('path/to/ai-config.json'); // 可以取消注释这行，如果你的配置在文件中 aiProviderRegistry.loadGlobalConfigs({ // 使用代码直接加载配置，或者从环境变量读取 'openai': { // 定义名为 'openai' 的 Provider 配置 apiKey: process.env.OPENAI_API_KEY, // 从环境变量安全地读取 API Key // defaultModel: 'gpt-4' // 可以设置默认模型 }, // 'gemini': { apiKey: process.env.GEMINI_API_KEY }, // 可以添加其他 Provider 配置 }); // 注意: 如果你的 Provider (比如自定义的本地模型 Provider) 需要注册，可以在这里调用 aiProviderRegistry.registerProvider(...) // 1.2 设置 FlowLab 核心组件 const logger = new ConsoleLogger('[AI-Flow]'); // 创建日志记录器，方便调试 const eventManager = new InMemoryEventManager(); // 创建内存事件管理器，用于节点间或工作流间通信（本例未使用，但大型应用可能需要） // 创建调度器，如果 AI 任务可能耗时较长，希望异步执行而不阻塞主流程时会用到 (需要配合 BaseAINode 的 scheduleTask 输入) const scheduler = new LocalScheduler(eventManager, logger); // 本地调度器，用于在同一进程中延迟或异步执行 const nodeRegistry = new NodeRegistry(); // 创建节点注册表，用于管理所有可用的节点类型 // --- 第 2 步: 注册所有需要在工作流中使用的节点 --- // 必须将所有会用到的节点类型（无论是 Core 节点还是 AI 节点）注册到 NodeRegistry 中 // 注册 Core 基础节点 nodeRegistry.register(new LogNode()); nodeRegistry.register(new HttpRequestNode()); // 假设这个节点用于调用外部 API // 注册 AI 节点实例 (如果节点需要调度器，请在构造时传入) nodeRegistry.register(new LLMCompletionNode(scheduler)); // 文本生成节点 nodeRegistry.register(new EmbeddingNode(scheduler)); // 向量嵌入节点 nodeRegistry.register(new FunctionCallingNode(scheduler)); // 函数调用决策节点 nodeRegistry.register(new VectorSearchNode()); // 向量搜索节点 (假设它不需要调度器) // --- 第 3 步: 定义包含 RAG 和函数调用的 AI 工作流 --- const ragWorkflow = new WorkflowDefinition('rag-function-call-flow', 'RAG问答与函数调用流程'); // 创建工作流定义，给它一个 ID 和名称 ragWorkflow // 使用链式调用定义工作流的步骤 // 起始步骤: 记录收到的用户查询 .addStep({ id: 'logStart', // 步骤的唯一标识符 nodeId: 'LogNode', // 使用哪个注册的节点类型来执行此步骤 input: { // 传递给 LogNode 的输入参数 // 使用模板字符串动态插入工作流的初始输入 'query' message: "RAG 流程启动。用户查询: ${workflow.input.query}" }, nextStepId: 'generateQueryEmbedding' // 定义下一个要执行的步骤的 ID }) // RAG 步骤 1: 为用户查询生成向量嵌入 (Embedding) .addStep({ id: 'generateQueryEmbedding', nodeId: 'EmbeddingNode', // 使用向量嵌入 AI 节点 input: { // EmbeddingNode 的输入参数 providerName: 'openai', // 指定使用哪个 AI Provider (需要已配置) text: '${workflow.input.query}', // 从工作流初始输入获取查询文本 modelOptions: { model: 'text-embedding-ada-002' } // 指定嵌入模型 (示例) }, outputMapping: { // 将节点的输出映射到工作流变量中，供后续步骤使用 // 将 EmbeddingNode 输出的 embedding 字段，存入名为 'queryEmbedding' 的工作流变量 'workflow.variables.queryEmbedding': 'output.embedding' }, nextStepId: 'searchVectorDB' // 下一步：进行向量搜索 }) // RAG 步骤 2: 使用查询向量在向量数据库中搜索相关文档 .addStep({ id: 'searchVectorDB', nodeId: 'VectorSearchNode', // 使用（假设存在的）向量搜索节点 input: { // 向量搜索节点的输入 queryEmbedding: '${workflow.variables.queryEmbedding}', // 使用上一步生成的查询向量 topK: 3, // 指定返回最相关的 3 个文档 // 此处通常还需要传递向量数据库的连接信息或配置 // vectorDbConfig: { host: '...', indexName: '...' } }, outputMapping: { // 将搜索结果 (假设是文档数组) 存入 'retrievedDocs' 工作流变量 'workflow.variables.retrievedDocs': 'output.documents' // 期望输出格式: [{content: '...', score: 0.9}, ...] }, nextStepId: 'checkDocsFound' // 下一步：检查是否找到了文档 }) // 条件步骤: 检查 RAG 步骤是否找到了相关文档 .addConditionStep({ id: 'checkDocsFound', condition: (context) => // 使用一个函数来评估条件，函数接收当前工作流上下文 // 检查 'retrievedDocs' 变量是否存在且包含元素 (context.variables.retrievedDocs && context.variables.retrievedDocs.length > 0), branches: { // 定义不同条件结果对应的下一个步骤 ID 'true': 'synthesizeAnswer', // 如果条件为 true (找到文档)，则跳转到 'synthesizeAnswer' 'false': 'handleNoDocs' // 如果条件为 false (未找到文档)，则跳转到 'handleNoDocs' } // 注意：如果 condition 函数需要异步操作（例如调用 AI），需要使用 async 并确保返回 Promise // condition: async (context) => { ... } }) // RAG 步骤 3a (路径: 找到文档): 使用 LLM 结合用户查询和检索到的文档来生成答案 .addStep({ id: 'synthesizeAnswer', nodeId: 'LLMCompletionNode', // 使用文本生成 AI 节点 input: { // 文本生成节点的输入 providerName: 'openai', // 指定 AI Provider modelOptions: { model: 'gpt-4', temperature: 0.5 }, // 指定模型和参数 // 核心: 构建 Prompt，包含原始查询和检索到的上下文信息 promptTemplate: `用户查询: \${workflow.input.query}\n\n相关上下文:\n${'${workflow.variables.retrievedDocs.map(d => `- ${d.content}`).join("\\n")}'}\n\n请*仅*根据上面提供的上下文来回答用户查询。如果上下文中没有答案，请明确说明。`, // 注意: 上面模板中使用了 JS 模板字符串来动态构建上下文部分。 // 对于复杂的模板处理，可能需要在专门的脚本节点中预处理，或者使用更强大的模板引擎。 // 这里假设 FlowLab 的模板渲染能处理简单的 JS 表达式（如果不能，需要调整）。 }, outputMapping: { // 将 LLM 生成的答案文本存入 'synthesizedAnswer' 工作流变量 'workflow.variables.synthesizedAnswer': 'output.completionText' }, nextStepId: 'decideNextAction' // 下一步：让 AI 决定是否需要额外操作 }) // RAG 步骤 3b (路径: 未找到文档): 处理未找到相关文档的情况 .addStep({ id: 'handleNoDocs', nodeId: 'LogNode', // 可以仅记录日志，或设置一个默认回答 input: { message: "未能根据查询找到相关文档。" }, outputMapping: { // 设置一个默认的、表示未找到答案的文本到 'synthesizedAnswer' 变量 'workflow.variables.synthesizedAnswer': '"抱歉，我无法在知识库中找到您查询的相关信息。"' }, nextStepId: 'decideNextAction' // 即使没找到文档，也继续下一步判断是否需要其他操作 }) // 函数调用步骤 1: 使用 Function Calling 节点让 AI 决定是否需要调用外部函数 (如下单、创建工单等) .addStep({ id: 'decideNextAction', nodeId: 'FunctionCallingNode', // 使用函数调用决策 AI 节点 input: { // 函数调用节点的输入 providerName: 'openai', // 指定 AI Provider modelOptions: { model: 'gpt-4' }, // 指定支持函数调用的模型 // Prompt 指导 AI 基于查询和已生成的答案，判断是否需要创建工单 promptTemplate: `根据用户原始查询 "\${workflow.input.query}" 以及系统提供的答案 "\${workflow.variables.synthesizedAnswer}"，请判断是否需要为用户创建一个支持工单。仅当用户看起来不满意、问题未解决或需要进一步协助时才调用函数。`, // 定义可供 AI 选择调用的函数列表 (遵循特定格式，如 OpenAI Function Calling API) functions: [ { name: 'create_support_ticket', // 函数名称 description: '为用户创建一个支持工单。', // 函数描述，供 AI 理解其作用 parameters: { // 函数参数的 JSON Schema 定义 type: 'object', properties: { summary: { type: 'string', description: '用户问题的简短摘要。' }, query: { type: 'string', description: '用户的原始查询文本。' } }, required: ['summary', 'query'] // 必须提供的参数 } } // 可以定义更多函数供 AI 选择 ] }, outputMapping: { // 将 FunctionCallingNode 的整个输出 (包含 functionCall 或 responseText) 存入 'actionDecision' 变量 'workflow.variables.actionDecision': 'output' // 输出结构: { functionCall?: { name: string, arguments: object }, responseText?: string, usage?: ... } }, nextStepId: 'checkFunctionCall' // 下一步：检查 AI 是否要求调用函数 }) // 条件步骤: 检查上一步 AI 是否决定调用函数 .addConditionStep({ id: 'checkFunctionCall', // 检查 'actionDecision' 变量中是否存在 functionCall 字段 condition: (context) => !!context.variables.actionDecision?.functionCall, branches: { 'true': 'callCreateTicketAPI', // 如果 AI 要求调用函数，则跳转到执行 API 调用步骤 'false': 'logFinalAnswer' // 如果 AI 未要求调用函数，则直接记录最终答案 } }) // 函数调用步骤 2a (路径: 需要调用函数): 实际执行外部 API 调用来创建工单 .addStep({ id: 'callCreateTicketAPI', nodeId: 'HttpRequestNode', // 使用通用的 HTTP 请求节点 input: { // HTTP 请求节点的输入 method: 'POST', // 请求方法 url: 'https://api.example.com/support/tickets', // 你的实际工单系统 API 地址 headers: { // 请求头，例如认证信息 'Authorization': 'Bearer YOUR_API_TOKEN', // 替换为你的 API Token 'Content-Type': 'application/json' }, // 从 AI 决定的函数调用的参数中，动态构建请求体 (body) body: { summary: '${workflow.variables.actionDecision.functionCall.arguments.summary}', // 获取 AI 提供的摘要 original_query: '${workflow.variables.actionDecision.functionCall.arguments.query}', // 获取 AI 提供的原始查询 context_answer: '${workflow.variables.synthesizedAnswer}' // 可以将 RAG 生成的答案也一并提交 // 注意: 模板替换需要能正确处理嵌套 JSON 路径 } }, outputMapping: { // 将 API 调用的结果存储起来 'workflow.variables.ticketResult': 'output.data' // 假设 HTTP 节点将响应体放在 output.data 中 }, nextStepId: 'logTicketCreated' // 下一步：记录工单创建结果 }) // 日志步骤 (路径: 工单已创建) .addStep({ id: 'logTicketCreated', nodeId: 'LogNode', input: { // 记录日志，包含 API 的响应信息 (需要确保 ticketResult 是可序列化的) message: "已请求创建支持工单。API 响应: ${JSON.stringify(workflow.variables.ticketResult ?? 'N/A')}" }, nextStepId: 'finalLog' // 跳转到最终日志步骤 }) // 函数调用步骤 2b (路径: 无需调用函数): 记录 AI 直接提供的最终文本答案 .addStep({ id: 'logFinalAnswer', nodeId: 'LogNode', input: { // 从 actionDecision 中获取 AI 的直接文本回复，如果不存在则使用 RAG 生成的答案 message: "最终答案 (无需创建工单): ${workflow.variables.actionDecision.responseText || workflow.variables.synthesizedAnswer}" }, nextStepId: 'finalLog' // 跳转到最终日志步骤 }) // 结束步骤: 记录整个流程完成 .addStep({ id: 'finalLog', nodeId: 'LogNode', input: { message: "RAG 与函数调用流程已完成。" } // 没有 nextStepId，表示这个分支的流程结束 }) // 最后，设置整个工作流的起始步骤 ID .setStartStep('logStart'); // --- 第 4 步: 执行工作流 --- async function runFlow() { // 4.1 定义工作流的初始输入数据 const initialInput = { query: "我的密码重置链接好像失效了，无法重置密码，怎么办？" }; // 4.2 创建本次执行的工作流上下文，传入工作流定义 ID 和初始输入 const context = createWorkflowContext(ragWorkflow.id, initialInput); // 4.3 创建工作流执行器实例 const executor = new WorkflowExecutor( ragWorkflow, // 要执行的工作流定义 context, // 本次执行的上下文 { // 执行器选项 nodeRegistry, // 必须提供节点注册表 eventManager, // 提供事件管理器 (可选) scheduler, // 提供调度器 (可选, 仅当使用 scheduleTask 时需要) logger, // 提供日志记录器 (推荐) // maxLoopIterations: 100 // (可选) 防止无限循环的最大迭代次数 // 可以注入其他依赖项给节点使用，例如数据库连接池等 } ); // 4.4 运行工作流并处理结果 try { console.log("--- 开始执行工作流 ---"); const result = await executor.run(); // 异步执行工作流 // 工作流执行完毕后，输出结果 console.log(`\n--- 工作流执行结果 ---`); console.log(`最终状态: ${result.status}`); // 输出工作流的最终状态 (COMPLETED, FAILED) console.log(`最终工作流变量:`, JSON.stringify(result.variables, null, 2)); // 输出执行后所有的工作流变量 // (可选) 输出详细的执行历史记录，用于调试 // console.log(`\n--- 执行历史 ---`); // result.history.forEach(h => console.log( // `步骤: ${h.stepId} (节点: ${h.nodeId}) -> 状态: ${h.status} ` + // `[开始: ${h.startTime.toISOString()}${h.endTime ? ' - 结束: ' + h.endTime.toISOString() : ''}]` + // `${h.error ? ' 错误: ' + h.error.message : ''}` // )); } catch (error) { // 如果执行器在运行过程中抛出未捕获的异常 console.error(`工作流执行失败:`, error); } } // 启动执行 runFlow(); ```