#FlowGram Runtime 源码导读

// 工作流执行的核心方法
public async run(params: RunParams): Promise<RunResult> {
  const { schema, inputs, options } = params;

  // 创建工作流上下文
  const context = this.createContext(schema, inputs, options);

  try {
    // 初始化工作流
    await this.initialize(context);

    // 执行工作流
    await this.execute(context);

    // 获取工作流结果
    const result = await this.getResult(context);

    return {
      status: 'success',
      outputs: result
    };
  } catch (error) {
    // 错误处理
    return {
      status: 'fail',
      error: error.message
    };
  }
}

// 执行工作流
private async execute(context: IContext): Promise<void> {
  // 获取开始节点
  const startNode = context.workflow.getStartNode();

  // 从开始节点开始执行
  await this.executeNode({ context, node: startNode });

  // 等待所有节点执行完成
  await this.waitForCompletion(context);
}

// 执行节点
public async executeNode(params: { context: IContext; node: INode }): Promise<void> {
  const { context, node } = params;

  // 获取节点执行器
  const executor = this.getExecutor(node.type);

  // 准备节点输入
  const inputs = await this.prepareInputs(context, node);

  // 执行节点
  const result = await executor.execute({
    node,
    inputs,
    context
  });

  // 处理节点输出
  await this.processOutputs(context, node, result.outputs);

  // 调度下一个节点
  await this.scheduleNextNodes(context, node);
}

export interface INodeExecutor {
  type: string;
  execute(context: ExecutionContext): Promise<ExecutionResult>;
}

export class LLMExecutor implements INodeExecutor {
  public type = 'llm';

  public async execute(context: ExecutionContext): Promise<ExecutionResult> {
    const inputs = context.inputs as LLMExecutorInputs;

    // 创建 LLM 提供商
    const provider = this.createProvider(inputs);

    // 准备提示词
    const systemPrompt = inputs.systemPrompt || '';
    const userPrompt = inputs.prompt || '';

    // 调用 LLM
    const result = await provider.call({
      systemPrompt,
      userPrompt,
      options: {
        temperature: inputs.temperature
      }
    });

    // 返回结果
    return {
      outputs: {
        result: result.content
      }
    };
  }

  private createProvider(inputs: LLMExecutorInputs): ILLMProvider {
    // 根据模型名称创建不同的提供商
    if (inputs.modelName.startsWith('gpt-')) {
      return new OpenAIProvider({
        apiKey: inputs.apiKey,
        apiHost: inputs.apiHost,
        modelName: inputs.modelName
      });
    }

    throw new Error(`Unsupported model: ${inputs.modelName}`);
  }
}

// 变量管理器
export class VariableManager implements IVariableManager {
  constructor(private store: IVariableStore) {}

  // 解析变量引用
  public async resolve(ref: ValueSchema, scope?: string): Promise<any> {
    if (ref.type === 'constant') {
      return ref.content;
    } else if (ref.type === 'ref') {
      const path = ref.content as string[];
      return this.get(path, scope);
    }
    throw new Error(`Unsupported value type: ${ref.type}`);
  }

  // 获取变量值
  public async get(path: string[], scope?: string): Promise<any> {
    const [nodeID, key, ...rest] = path;
    const value = await this.store.get(nodeID, key, scope);

    if (rest.length === 0) {
      return value;
    }

    // 处理嵌套属性
    return this.getNestedProperty(value, rest);
  }

  // 设置变量值
  public async set(nodeID: string, key: string, value: any, scope?: string): Promise<void> {
    await this.store.set(nodeID, key, value, scope);
  }
}

// 状态中心
export class StatusCenter implements IStatusCenter {
  private workflowStatus: Record<string, WorkflowStatus> = {};
  private nodeStatus: Record<string, Record<string, NodeStatus>> = {};

  // 设置工作流状态
  public setWorkflowStatus(workflowID: string, status: WorkflowStatus): void {
    this.workflowStatus[workflowID] = status;
  }

  // 获取工作流状态
  public getWorkflowStatus(workflowID: string): WorkflowStatus {
    return this.workflowStatus[workflowID] || 'idle';
  }

  // 设置节点状态
  public setNodeStatus(workflowID: string, nodeID: string, status: NodeStatus): void {
    if (!this.nodeStatus[workflowID]) {
      this.nodeStatus[workflowID] = {};
    }
    this.nodeStatus[workflowID][nodeID] = status;
  }

  // 获取节点状态
  public getNodeStatus(workflowID: string, nodeID: string): NodeStatus {
    return this.nodeStatus[workflowID]?.[nodeID] || 'idle';
  }
}

// 节点执行器工厂
export class NodeExecutorFactory implements INodeExecutorFactory {
  private executors: Record<string, INodeExecutor> = {};

  // 注册节点执行器
  public register(executor: INodeExecutor): void {
    this.executors[executor.type] = executor;
  }

  // 创建节点执行器
  public create(type: string): INodeExecutor {
    const executor = this.executors[type];
    if (!executor) {
      throw new Error(`No executor registered for node type: ${type}`);
    }
    return executor;
  }
}

// 节点执行器接口（策略接口）
export interface INodeExecutor {
  type: string;
  execute(context: ExecutionContext): Promise<ExecutionResult>;
}

// 具体策略实现
export class StartExecutor implements INodeExecutor {
  public type = 'start';

  public async execute(context: ExecutionContext): Promise<ExecutionResult> {
    // Start 节点的执行逻辑
  }
}

export class EndExecutor implements INodeExecutor {
  public type = 'end';

  public async execute(context: ExecutionContext): Promise<ExecutionResult> {
    // End 节点的执行逻辑
  }
}

// 事件发布者
export class EventEmitter implements IEventEmitter {
  private listeners: Record<string, Function[]> = {};

  // 订阅事件
  public on(event: string, listener: Function): void {
    if (!this.listeners[event]) {
      this.listeners[event] = [];
    }
    this.listeners[event].push(listener);
  }

  // 发布事件
  public emit(event: string, ...args: any[]): void {
    const eventListeners = this.listeners[event];
    if (eventListeners) {
      for (const listener of eventListeners) {
        listener(...args);
      }
    }
  }
}

// 依赖注入容器
export class Container {
  private static _instance: Container;
  private registry: Map<any, any> = new Map();

  public static get instance(): Container {
    if (!Container._instance) {
      Container._instance = new Container();
    }
    return Container._instance;
  }

  // 注册服务
  public register<T>(token: any, instance: T): void {
    this.registry.set(token, instance);
  }

  // 获取服务
  public resolve<T>(token: any): T {
    const instance = this.registry.get(token);
    if (!instance) {
      throw new Error(`No instance registered for token: ${token}`);
    }
    return instance;
  }
}