nodes/AIDC-AI-ComfyUI-Copilot
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: 解决/prompt api调用问题

Browse Source
This commit is contained in:
ai-business-hql
2025-07-29 15:25:57 +08:00
parent 19662fc9ec
commit cc2d69beab
2 changed files with 105 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/service/debug_agent.py
+21 -87
View File
@@ -3,6 +3,7 @@ Debug Agent for ComfyUI Workflow Error Analysis
'''
import os
import json
import requests
from typing import List, Dict, Any, Optional
from agents._config import set_default_openai_api
from agents.agent import Agent
@@ -39,103 +40,36 @@ load_env_config()
set_default_openai_api("chat_completions")
set_tracing_disabled(True)
# @function_tool
# def run_workflow(session_id: str) -> str:
# """验证当前session的工作流并返回结果"""
# try:
# workflow_data = get_workflow_data(session_id)
# if not workflow_data:
# return json.dumps({"error": "No workflow data found for this session"})
# print(f"Run workflow for session {session_id}")
# # 使用 ComfyGateway 调用 server.py 的 post_prompt 逻辑
# from ..utils.comfy_gateway import ComfyGateway
# gateway = ComfyGateway()
# # 准备请求数据格式（与server.py post_prompt接口一致）
# request_data = {
# "prompt": workflow_data,
# "client_id": f"debug_agent_{session_id}"
# }
# # 调用ComfyGateway.run_prompt，它会直接复用server.py的post_prompt逻辑
# result = gateway.run_prompt(request_data)
# return json.dumps(result)
# except Exception as e:
# return json.dumps({"error": f"Failed to run workflow: {str(e)}"})
@function_tool
def run_workflow(session_id: str) -> str:
async def run_workflow(session_id: str) -> str:
"""验证当前session的工作流并返回结果"""
try:
workflow_data = get_workflow_data(session_id)
if not workflow_data:
return json.dumps({"error": "No workflow data found for this session"})
print(f"Validating workflow for session {session_id}")
print(f"Run workflow for session {session_id}")
# 直接调用 ComfyUI 的内部验证函数，避免 HTTP 请求
try:
# 调用工作流验证
valid = execution.validate_prompt(workflow_data)
if valid[0]: # 验证成功
# valid = (True, None, outputs_to_execute, node_errors)
prompt_id = str(uuid.uuid4())
result = {
"success": True,
"prompt_id": prompt_id,
"outputs_to_execute": valid[2],
"node_errors": valid[3],
"message": "Workflow validation successful"
}
print(f"Workflow validation successful for session {session_id}")
return json.dumps(result)
else:
# valid = (False, error, outputs_to_execute, node_errors)
error_info = valid[1] # 错误信息
node_errors = valid[3] # 节点错误
result = {
"success": False,
"error": error_info,
"node_errors": node_errors,
"message": "Workflow validation failed"
}
print(f"Workflow validation failed for session {session_id}: {error_info}")
return json.dumps(result)
except Exception as e:
print(f"Exception during workflow validation: {str(e)}")
import traceback
traceback.print_exc()
return json.dumps({
"success": False,
"error": {
"type": "validation_exception",
"message": f"Exception during validation: {str(e)}",
"details": str(e)
},
"node_errors": {}
})
# 使用 ComfyGateway 调用 server.py 的 post_prompt 逻辑
from ..utils.comfy_gateway import ComfyGateway
# 简化方法：直接使用 requests 同步调用
gateway = ComfyGateway()
# 准备请求数据格式（与server.py post_prompt接口一致）
request_data = {
"prompt": workflow_data,
"client_id": f"debug_agent_{session_id}"
}
result = await gateway.run_prompt(request_data)
print(result)
return json.dumps(result)
except Exception as e:
print(f"Error in run_workflow: {str(e)}")
import traceback
traceback.print_exc()
return json.dumps({
"success": False,
"error": {
"type": "function_error",
"message": f"Failed to validate workflow: {str(e)}",
"details": str(e)
},
"node_errors": {}
})
return json.dumps({"error": f"Failed to run workflow: {str(e)}"})
@function_tool
def analyze_error_type(error_data: str) -> str:
backend/utils/comfy_gateway.py
+84 -80
View File
@@ -2,13 +2,14 @@
ComfyUI Gateway Utilities
This module provides Python implementations of ComfyUI API functions,
directly calling ComfyUI internal functions instead of HTTP requests.
using HTTP requests to the ComfyUI server for consistency.
"""
import json
import os
import uuid
import logging
import asyncio
from typing import Dict, Any, Optional, List
# Import ComfyUI internal modules
@@ -16,6 +17,7 @@ import nodes
import execution
import folder_paths
import server
import aiohttp
class ComfyGateway:
@@ -26,20 +28,35 @@ class ComfyGateway:
Initialize ComfyUI Gateway
Args:
base_url: Deprecated - no longer used since we call internal functions directly
base_url: Optional base URL for ComfyUI server. If not provided, will auto-detect.
"""
if base_url:
logging.warning("ComfyGateway: base_url parameter is deprecated when using internal calls")
# Get server instance for operations that need it
self.server_instance = server.PromptServer.instance
def run_prompt(self, json_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Run a prompt - direct internal call equivalent of server.py post_prompt
This method replicates the exact logic from server.py's post_prompt route handler
to ensure consistent behavior without requiring HTTP requests.
# Auto-detect server URL if not provided
if base_url:
self.base_url = base_url.rstrip('/')
else:
# Auto-detect from server instance
if hasattr(self.server_instance, 'address') and hasattr(self.server_instance, 'port'):
address = self.server_instance.address or '127.0.0.1'
port = self.server_instance.port or 8188
# Use 127.0.0.1 for localhost to avoid potential connection issues
if address in ['0.0.0.0', '::']:
address = '127.0.0.1'
self.base_url = f"http://{address}:{port}"
else:
# Fallback to default
self.base_url = "http://127.0.0.1:8188"
logging.info(f"ComfyGateway initialized with base_url: {self.base_url}")
async def run_prompt(self, json_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Run a prompt - HTTP call to ComfyUI /api/prompt endpoint
This method sends an HTTP POST request to the ComfyUI server's /api/prompt endpoint
to ensure consistent behavior and avoid code duplication.
Args:
json_data: The prompt/workflow data in the same format as HTTP API
@@ -48,68 +65,58 @@ class ComfyGateway:
Dict containing the validation result, similar to HTTP API response
"""
try:
# Trigger on_prompt handlers (same as server.py)
json_data = self.server_instance.trigger_on_prompt(json_data)
print(json_data)
# Handle number logic (same as server.py)
if "number" in json_data:
number = float(json_data['number'])
else:
number = self.server_instance.number
if "front" in json_data:
if json_data['front']:
number = -number
self.server_instance.number += 1
# Create a timeout configuration
timeout = aiohttp.ClientTimeout(total=30) # 30 second timeout
# Make HTTP request to /api/prompt endpoint
url = f"{self.base_url}/api/prompt"
headers = {
'Content-Type': 'application/json'
}
# Main prompt validation and processing logic (same as server.py)
if "prompt" in json_data:
prompt = json_data["prompt"]
valid = execution.validate_prompt(prompt)
extra_data = {}
if "extra_data" in json_data:
extra_data = json_data["extra_data"]
if "client_id" in json_data:
extra_data["client_id"] = json_data["client_id"]
if valid[0]:
# Validation successful
prompt_id = str(uuid.uuid4())
outputs_to_execute = valid[2]
# Add to prompt queue (same as server.py)
if hasattr(self.server_instance, 'prompt_queue') and self.server_instance.prompt_queue:
self.server_instance.prompt_queue.put((number, prompt_id, prompt, extra_data, outputs_to_execute))
response = {
"success": True,
"prompt_id": prompt_id,
"number": number,
"node_errors": valid[3]
}
return response
else:
# Validation failed
logging.warning("invalid prompt: {}".format(valid[1]))
return {
"success": False,
"error": valid[1],
"node_errors": valid[3]
}
else:
# No prompt provided
error = {
"type": "no_prompt",
"message": "No prompt provided",
"details": "No prompt provided",
"extra_info": {}
# Create temporary session
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.post(url, json=json_data, headers=headers) as response:
response_data = await response.json()
status_code = response.status
# Handle the response based on status code
if status_code == 200:
# Success response - add success flag for consistency
return {
"success": True,
**response_data
}
else:
# Error response (400, etc.) - add success flag for consistency
return {
"success": False,
"error": error,
"node_errors": {}
**response_data
}
except aiohttp.ClientConnectionError as e:
logging.error(f"Connection error in run_prompt: {e}")
return {
"success": False,
"error": {
"type": "connection_error",
"message": f"Failed to connect to ComfyUI server at {self.base_url}",
"details": str(e)
},
"node_errors": {}
}
except aiohttp.ClientTimeout as e:
logging.error(f"Timeout error in run_prompt: {e}")
return {
"success": False,
"error": {
"type": "timeout_error",
"message": "Request to ComfyUI server timed out",
"details": str(e)
},
"node_errors": {}
}
except Exception as e:
logging.error(f"Error in run_prompt: {e}")
return {
@@ -260,47 +267,41 @@ class ComfyGateway:
# Convenience functions for backward compatibility and easy importing
def run_prompt(json_data: Dict[str, Any], base_url: Optional[str] = None) -> Dict[str, Any]:
async def run_prompt(json_data: Dict[str, Any], base_url: Optional[str] = None) -> Dict[str, Any]:
"""
Standalone function to run a prompt - direct equivalent of TypeScript runPrompt()
Standalone function to run a prompt - HTTP call to ComfyUI /api/prompt endpoint
Args:
json_data: The prompt/workflow data to execute
base_url: Deprecated - no longer used
base_url: Optional base URL for ComfyUI server
Returns:
Dict containing the API response
"""
gateway = ComfyGateway(base_url)
return gateway.run_prompt(json_data)
return await gateway.run_prompt(json_data)
def get_object_info(base_url: Optional[str] = None) -> Dict[str, Any]:
"""Standalone function to get object info - direct internal call equivalent"""
if base_url:
logging.warning("get_object_info: base_url parameter is deprecated when using internal calls")
gateway = ComfyGateway(base_url)
return gateway.get_object_info()
def get_object_info_by_class(node_class: str, base_url: Optional[str] = None) -> Dict[str, Any]:
"""Standalone function to get object info for specific node class - direct internal call equivalent"""
if base_url:
logging.warning("get_object_info_by_class: base_url parameter is deprecated when using internal calls")
gateway = ComfyGateway(base_url)
return gateway.get_object_info(node_class)
def get_installed_nodes(base_url: Optional[str] = None) -> List[str]:
"""Standalone function to get installed nodes - direct internal call equivalent"""
if base_url:
logging.warning("get_installed_nodes: base_url parameter is deprecated when using internal calls")
gateway = ComfyGateway(base_url)
return gateway.get_installed_nodes()
# Example usage
if __name__ == "__main__":
# Example of how to use the ComfyGateway with internal calls
async def main():
# Example of how to use the ComfyGateway with HTTP calls
gateway = ComfyGateway()
# Example prompt structure (you would replace this with actual workflow data)
@@ -322,12 +323,15 @@ if __name__ == "__main__":
print(f"Found {len(nodes_list)} installed nodes")
# Get object info by class
node_info = gateway.get_object_info_by_class("CLIPTextEncode")
node_info = gateway.get_object_info("CLIPTextEncode")
print(f"Node info: {node_info}")
# Validate a prompt (uncomment to actually execute)
# result = gateway.run_prompt(example_prompt)
# result = await gateway.run_prompt(example_prompt)
# print(f"Prompt result: {result}")
except Exception as e:
print(f"Example failed: {e}")
if __name__ == "__main__":
asyncio.run(main())