Xiao-zhen-Liu commented on code in PR #4020: URL: https://github.com/apache/texera/pull/4020#discussion_r2535013066
########## frontend/src/app/workspace/service/copilot/texera-copilot.ts: ########## @@ -0,0 +1,388 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { Injectable } from "@angular/core"; +import { BehaviorSubject, Observable, from, of, throwError, defer } from "rxjs"; +import { map, catchError, tap, switchMap, finalize } from "rxjs/operators"; +import { WorkflowActionService } from "../workflow-graph/model/workflow-action.service"; +import { toolWithTimeout } from "./tool/tools-utility"; +import * as CurrentWorkflowTools from "./tool/current-workflow-editing-observing-tools"; +import * as MetadataTools from "./tool/workflow-metadata-tools"; +import { ToolOperatorAccess, parseOperatorAccessFromStep } from "./tool/react-step-operator-parser"; +import { OperatorMetadataService } from "../operator-metadata/operator-metadata.service"; +import { createOpenAI } from "@ai-sdk/openai"; +import { generateText, type ModelMessage, stepCountIs } from "ai"; +import { WorkflowUtilService } from "../workflow-graph/util/workflow-util.service"; +import { AppSettings } from "../../../common/app-setting"; +import { WorkflowCompilingService } from "../compile-workflow/workflow-compiling.service"; +import { COPILOT_SYSTEM_PROMPT } from "./copilot-prompts"; +import { NotificationService } from "../../../common/service/notification/notification.service"; + +export enum CopilotState { + UNAVAILABLE = "Unavailable", + AVAILABLE = "Available", + GENERATING = "Generating", + STOPPING = "Stopping", +} + +/** + * Represents a single step in the ReAct (Reasoning and Acting) conversation flow. + * Each step can be either a user message or an agent response with potential tool calls. + */ +export interface ReActStep { + messageId: string; + stepId: number; + role: "user" | "agent"; + content: string; + isBegin: boolean; + isEnd: boolean; + timestamp: number; + toolCalls?: any[]; + toolResults?: any[]; + usage?: { + inputTokens?: number; + outputTokens?: number; + totalTokens?: number; + cachedInputTokens?: number; + }; + /** + * Map from tool call index to operator access information, which tracks operators were viewed or modified during the tool call. + */ + operatorAccess?: Map<number, ToolOperatorAccess>; +} + +/** + * Texera Copilot Service provides AI-powered assistance for workflow creation and manipulation. + * + * This service manages a single AI agent instance that can: + * 1. Interact with users through natural language messages + * 2. Execute workflow operations using specialized tools + * 3. Maintain conversation history and state + * + * The service communicates with an LLM backend (via LiteLLM) to generate responses and uses + * workflow tools to perform actions like listing operators, getting operator schemas, and + * manipulating workflow components. + * + * State management includes: + * - UNAVAILABLE: Agent not initialized + * - AVAILABLE: Agent ready to receive messages + * - GENERATING: Agent currently processing and generating response + * - STOPPING: Agent in the process of stopping generation + */ +@Injectable() +export class TexeraCopilot { + /** + * Maximum number of ReAct reasoning/action cycles allowed per generation. + * Prevents infinite loops and excessive token usage. + */ + private static readonly MAX_REACT_STEPS = 50; + + private model: any; + private modelType = ""; + private agentName = ""; + + /** + * Conversation history in LLM API format. + * Used internally to maintain context for generateText() API calls. + * Contains the raw message format expected by the AI model. + */ + private messages: ModelMessage[] = []; + + /** + * Representing a step in ReAct (Reasoning + Acting). + * This is what gets displayed in the UI to show the agent's reasoning process. + * Each step contains messageId (randomly generated UUID) and stepId (incremental from 0). + */ + private reActSteps: ReActStep[] = []; + private reActStepsSubject = new BehaviorSubject<ReActStep[]>([]); + public reActSteps$ = this.reActStepsSubject.asObservable(); + + private state = CopilotState.UNAVAILABLE; + private stateSubject = new BehaviorSubject<CopilotState>(CopilotState.UNAVAILABLE); + public state$ = this.stateSubject.asObservable(); + private tools: Record<string, any> = {}; + + constructor( + private workflowActionService: WorkflowActionService, + private workflowUtilService: WorkflowUtilService, + private operatorMetadataService: OperatorMetadataService, + private workflowCompilingService: WorkflowCompilingService, + private notificationService: NotificationService + ) {} + + public setAgentInfo(agentName: string): void { + this.agentName = agentName; + } + + public setModelType(modelType: string): void { + this.modelType = modelType; + } + + private setState(newState: CopilotState): void { + this.state = newState; + this.stateSubject.next(newState); + } + + private emitReActStep( + messageId: string, + stepId: number, + role: "user" | "agent", + content: string, + isBegin: boolean, + isEnd: boolean, + toolCalls?: any[], + toolResults?: any[], + usage?: ReActStep["usage"], + operatorAccess?: Map<number, ToolOperatorAccess> + ): void { + this.reActSteps.push({ + messageId, + stepId, + role, + content, + isBegin, + isEnd, + timestamp: Date.now(), + toolCalls, + toolResults, + usage, + operatorAccess, + }); + this.reActStepsSubject.next([...this.reActSteps]); + } + + public initialize(): Observable<void> { + return defer(() => { + try { + this.model = createOpenAI({ + baseURL: new URL(`${AppSettings.getApiEndpoint()}`, document.baseURI).toString(), + // apiKey is required by the library for creating the OpenAI compatible client; + // For security reason, we store the apiKey at the backend, thus the value is dummy here. + apiKey: "dummy", + }).chat(this.modelType); + + // Create tools once during initialization + this.tools = this.createWorkflowTools(); + + this.setState(CopilotState.AVAILABLE); + return of(undefined); + } catch (error: unknown) { + this.setState(CopilotState.UNAVAILABLE); + return throwError(() => error); + } + }); + } + + public sendMessage(message: string): Observable<void> { + return defer(() => { + if (!this.model) { + return throwError(() => new Error("Copilot not initialized")); + } + + if (this.state !== CopilotState.AVAILABLE) { + return throwError(() => new Error(`Cannot send message: agent is ${this.state}`)); + } + + this.setState(CopilotState.GENERATING); + + // Generate unique message ID for this conversation turn + const messageId = crypto.randomUUID(); + let stepId = 0; + + // Emit user message as first step + this.emitReActStep(messageId, stepId++, "user", message, true, true); + this.messages.push({ role: "user", content: message }); + + let isFirstStep = true; + + /** + * Generate text using the AI model with ReAct (Reasoning + Acting) pattern. + * This is the core of the agent lifecycle with several callbacks: + * + * Lifecycle flow: + * 1. generateText() starts the LLM generation + * 2. stopWhen() - checked before each step to determine if generation should stop + * 3. onStepFinish() - called DURING generation after each reasoning/action step (real-time updates) + * 4. pipe operators - executed AFTER generation completes (final processing) + */ + return from( + generateText({ + model: this.model, + messages: this.messages, + tools: this.tools, + system: COPILOT_SYSTEM_PROMPT, + /** + * stopWhen - Determines if generation should stop. + * Called before each step during generation. + * Returns true to stop, false to continue. + */ + stopWhen: ({ steps }) => { + if (this.state === CopilotState.STOPPING) { + this.notificationService.info(`Agent ${this.agentName} has stopped generation`); + return true; + } + // Stop if step count reaches max limit to prevent infinite loops + return stepCountIs(TexeraCopilot.MAX_REACT_STEPS)({ steps }); + }, + /** + * onStepFinish is called DURING generation after each ReAct step completes. + * This provides real-time updates to the UI as the agent reasons and acts. + * + * Each step may include: + * - text: The agent's reasoning or response text + * - toolCalls: Tools the agent decided to call + * - toolResults: Results from executed tools + * - usage: Token usage for this step + * + * Note: This is called multiple times during a single generation, + * once per reasoning/action cycle. + */ + onStepFinish: ({ text, toolCalls, toolResults, usage }) => { + if (this.state === CopilotState.STOPPING) { + return; + } + + // Parse operator access from tool results to track viewed/modified operators + const operatorAccess = parseOperatorAccessFromStep(toolCalls || [], toolResults || []); Review Comment: This PR would not have actions that modify operators, right? ########## frontend/src/styles.scss: ########## @@ -20,7 +20,7 @@ @import "@ali-hm/angular-tree-component/css/angular-tree-component.css"; * { - user-select: none; + user-select: text; Review Comment: This is an important change. Please mention it in the PR description. ########## frontend/src/app/workspace/component/workflow-editor/context-menu/context-menu/context-menu.component.html: ########## @@ -139,7 +139,7 @@ <!-- Delete option for links only --> <li nz-menu-item - *ngIf="hasHighlightedLinks() && + *ngIf="hasHighlightedLinks() && Review Comment: The changes are still there ########## frontend/src/app/workspace/component/agent-panel/agent-chat/agent-chat.component.ts: ########## @@ -0,0 +1,236 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { Component, ViewChild, ElementRef, Input, OnInit, AfterViewChecked } from "@angular/core"; +import { UntilDestroy, untilDestroyed } from "@ngneat/until-destroy"; +import { CopilotState, AgentUIMessage } from "../../../service/copilot/texera-copilot"; +import { AgentInfo, TexeraCopilotManagerService } from "../../../service/copilot/texera-copilot-manager.service"; +import { NotificationService } from "../../../../common/service/notification/notification.service"; + +@UntilDestroy() +@Component({ + selector: "texera-agent-chat", + templateUrl: "agent-chat.component.html", + styleUrls: ["agent-chat.component.scss"], +}) +export class AgentChatComponent implements OnInit, AfterViewChecked { + @Input() agentInfo!: AgentInfo; + @ViewChild("messageContainer", { static: false }) messageContainer?: ElementRef; + @ViewChild("messageInput", { static: false }) messageInput?: ElementRef; + + public agentResponses: AgentUIMessage[] = []; + public currentMessage = ""; + private shouldScrollToBottom = false; + public isDetailsModalVisible = false; + public selectedResponse: AgentUIMessage | null = null; + public hoveredMessageIndex: number | null = null; + public isSystemInfoModalVisible = false; + public systemPrompt: string = ""; + public availableTools: Array<{ name: string; description: string; inputSchema: any }> = []; + public agentState: CopilotState = CopilotState.UNAVAILABLE; + + constructor( + private copilotManagerService: TexeraCopilotManagerService, + private notificationService: NotificationService + ) {} + + ngOnInit(): void { + if (!this.agentInfo) { + return; + } + + // Subscribe to agent responses + this.copilotManagerService + .getAgentResponsesObservable(this.agentInfo.id) + .pipe(untilDestroyed(this)) + .subscribe(responses => { + this.agentResponses = responses; + this.shouldScrollToBottom = true; + }); + + // Subscribe to agent state changes + this.copilotManagerService + .getAgentStateObservable(this.agentInfo.id) + .pipe(untilDestroyed(this)) + .subscribe(state => { + this.agentState = state; + }); + } + + ngAfterViewChecked(): void { + if (this.shouldScrollToBottom) { + this.scrollToBottom(); + this.shouldScrollToBottom = false; + } + } + + public setHoveredMessage(index: number | null): void { + this.hoveredMessageIndex = index; + } + + public showResponseDetails(response: AgentUIMessage): void { + this.selectedResponse = response; + this.isDetailsModalVisible = true; + } + + public closeDetailsModal(): void { + this.isDetailsModalVisible = false; + this.selectedResponse = null; + } + + public showSystemInfo(): void { + this.copilotManagerService + .getSystemInfo(this.agentInfo.id) + .pipe(untilDestroyed(this)) + .subscribe(systemInfo => { + this.systemPrompt = systemInfo.systemPrompt; + this.availableTools = systemInfo.tools; + this.isSystemInfoModalVisible = true; + }); + } + + public closeSystemInfoModal(): void { + this.isSystemInfoModalVisible = false; + } + + public formatJson(data: any): string { + return JSON.stringify(data, null, 2); + } + + public getToolResult(response: AgentUIMessage, toolCallIndex: number): any { + if (!response.toolResults || toolCallIndex >= response.toolResults.length) { + return null; + } + const toolResult = response.toolResults[toolCallIndex]; + return toolResult.output || toolResult.result || toolResult; + } + + public getTotalInputTokens(): number { + for (let i = this.agentResponses.length - 1; i >= 0; i--) { + const response = this.agentResponses[i]; + if (response.usage?.inputTokens !== undefined) { + return response.usage.inputTokens; + } + } + return 0; + } + + public getTotalOutputTokens(): number { + for (let i = this.agentResponses.length - 1; i >= 0; i--) { + const response = this.agentResponses[i]; + if (response.usage?.outputTokens !== undefined) { + return response.usage.outputTokens; + } + } + return 0; + } + + /** + * Send a message to the agent via the copilot manager service. + */ + public sendMessage(): void { + if (!this.currentMessage.trim() || !this.canSendMessage()) { + return; + } + + const userMessage = this.currentMessage.trim(); + this.currentMessage = ""; + + // Send to copilot via manager service + this.copilotManagerService + .sendMessage(this.agentInfo.id, userMessage) + .pipe(untilDestroyed(this)) + .subscribe({ + error: (error: unknown) => { + this.notificationService.error(`Error sending message: ${error}`); + }, + }); + } + + /** + * Check if messages can be sent (only when agent is available). + */ + public canSendMessage(): boolean { + return this.agentState === CopilotState.AVAILABLE; + } + + /** + * Get the state icon URL based on current agent state. + * Uses the same icons as workflow operators for consistency. + */ + public getStateIconUrl(): string { + return this.agentState === CopilotState.AVAILABLE ? "assets/svg/done.svg" : "assets/gif/loading.gif"; + } + + /** + * Get the tooltip text for the state icon. + */ + public getStateTooltip(): string { + switch (this.agentState) { + case CopilotState.AVAILABLE: + return "Agent is ready"; + case CopilotState.GENERATING: + return "Agent is generating response..."; + case CopilotState.STOPPING: + return "Agent is stopping..."; + case CopilotState.UNAVAILABLE: + return "Agent is unavailable"; + default: + return "Agent status unknown"; + } + } + + public onEnterPress(event: KeyboardEvent): void { + if (!event.shiftKey) { + event.preventDefault(); + this.sendMessage(); + } + } + + private scrollToBottom(): void { + if (this.messageContainer) { + const element = this.messageContainer.nativeElement; + element.scrollTop = element.scrollHeight; + } + } + + public stopGeneration(): void { + this.copilotManagerService.stopGeneration(this.agentInfo.id).pipe(untilDestroyed(this)).subscribe(); + } + + public clearMessages(): void { + this.copilotManagerService.clearMessages(this.agentInfo.id).pipe(untilDestroyed(this)).subscribe(); + } + + public isGenerating(): boolean { + return this.agentState === CopilotState.GENERATING; + } + + public isStopping(): boolean { Review Comment: It's still there. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
