Skip to content

Commit

Permalink
Merge pull request #129 from djwu563/develop/v0.2.1
Browse files Browse the repository at this point in the history 
Add client:programmatic, and add handling for Ctrl+C exception
  • Loading branch information
@panregedit
panregedit authored Dec 17, 2024
2 parents aee3e80 + b3f3359 commit 95b0a01
Show file tree
Hide file tree
Showing 4 changed files with 289 additions and 170 deletions.
28 changes: 18 additions & 10 deletions omagent-core/src/omagent_core/clients/devices/app/client.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,8 @@
from omagent_core.utils.container import container
from omagent_core.utils.registry import registry
from omagent_core.services.connectors.redis import RedisConnector

from omagent_core.engine.http.models.workflow_status import terminal_status
from omagent_core.utils.logger import logging
registry.import_module()

container.register_connector(name='redis_stream_client', connector=RedisConnector)
Expand Down Expand Up @@ -41,16 +42,23 @@ def stop_interactor(self):
self._task_handler_interactor.stop_processes()

def start_processor(self):
worker_config = build_from_file(self._config_path)
self._task_handler_processor = TaskHandler(worker_config=worker_config, workers=self._workers)
self._task_handler_processor.start_processes()
workflow_instance_id = self._processor.start_workflow_with_input(workflow_input={})
while True:
status = self._processor.get_workflow(workflow_id=workflow_instance_id).status
if status == 'COMPLETED':
workflow_instance_id = self._processor.start_workflow_with_input(workflow_input={})
workflow_instance_id = None
try:
worker_config = build_from_file(self._config_path)
self._task_handler_processor = TaskHandler(worker_config=worker_config, workers=self._workers)
self._task_handler_processor.start_processes()
workflow_instance_id = self._processor.start_workflow_with_input(workflow_input={})
while True:
status = self._processor.get_workflow(workflow_id=workflow_instance_id).status
if status in terminal_status:
workflow_instance_id = self._processor.start_workflow_with_input(workflow_input={})

sleep(1)
sleep(1)
except KeyboardInterrupt:
logging.info("\nDetected Ctrl+C, stopping workflow...")
if workflow_instance_id is not None:
self._processor._executor.terminate(workflow_id=workflow_instance_id)
raise

def stop_processor(self):
self._task_handler_processor.stop_processes()
Expand Down
255 changes: 134 additions & 121 deletions omagent-core/src/omagent_core/clients/devices/cli/client.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from omagent_core.utils.container import container
container.register_connector(name='redis_stream_client', connector=RedisConnector)
from omagent_core.engine.orkes.orkes_workflow_client import workflow_client
from omagent_core.engine.http.models.workflow_status import terminal_status
from omagent_core.engine.workflow.conductor_workflow import ConductorWorkflow
from omagent_core.utils.build import build_from_file
from omagent_core.engine.automator.task_handler import TaskHandler
Expand Down Expand Up @@ -42,139 +43,151 @@ def __init__(
self._input_prompt = input_prompt

def start_interactor(self):
absolute_path = Path(self._config_path).resolve()
print(f"{absolute_path}")
worker_config = build_from_file(self._config_path)
print("worker_config:",worker_config)
self._task_handler_interactor = TaskHandler(worker_config=worker_config, workers=self._workers)
self._task_handler_interactor.start_processes()
workflow_instance_id = self._interactor.start_workflow_with_input(workflow_input={})

stream_name = f"{workflow_instance_id}_output"
consumer_name = f"{workflow_instance_id}_agent" # consumer name
group_name = "omappagent" # replace with your consumer group name
poll_interval = 1

if self._input_prompt:
self.first_input(workflow_instance_id=workflow_instance_id, input_prompt=self._input_prompt)

workflow_instance_id = None
try:
container.get_connector('redis_stream_client')._client.xgroup_create(
stream_name, group_name, id="0", mkstream=True
)
except Exception as e:
logging.debug(f"Consumer group may already exist: {e}")
absolute_path = Path(self._config_path).resolve()
print(f"{absolute_path}")
worker_config = build_from_file(self._config_path)
print("worker_config:",worker_config)
self._task_handler_interactor = TaskHandler(worker_config=worker_config, workers=self._workers)
self._task_handler_interactor.start_processes()
workflow_instance_id = self._interactor.start_workflow_with_input(workflow_input={})


while True:
try:
status = self._interactor.get_workflow(workflow_id=workflow_instance_id).status #获取执行状态
if status == 'COMPLETED':
break
data_flag = False
content = None
# logging.info(f"Checking workflow status: {workflow_instance_id}")
workflow_status = workflow_client.get_workflow_status(workflow_instance_id)
if workflow_status.status not in running_status:
logging.info(f"Workflow {workflow_instance_id} is not running, exiting...")
break
stream_name = f"{workflow_instance_id}_output"
consumer_name = f"{workflow_instance_id}_agent" # consumer name
group_name = "omappagent" # replace with your consumer group name
poll_interval = 1

# read new messages from consumer group
messages = container.get_connector('redis_stream_client')._client.xreadgroup(
group_name, consumer_name, {stream_name: ">"}, count=1
if self._input_prompt:
self.first_input(workflow_instance_id=workflow_instance_id, input_prompt=self._input_prompt)

try:
container.get_connector('redis_stream_client')._client.xgroup_create(
stream_name, group_name, id="0", mkstream=True
)
# Convert byte data to string
messages = [
(stream, [(message_id, {k.decode('utf-8'): v.decode('utf-8') for k, v in message.items()}) for message_id, message in message_list])
for stream, message_list in messages
]
# logging.info(f"Messages: {messages}")

for stream, message_list in messages:
for message_id, message in message_list:
data_flag, content = self.process_message(message)
# confirm message has been processed
container.get_connector('redis_stream_client')._client.xack(
stream_name, group_name, message_id
)
if data_flag:
contents = []
while True:
print(f"{Fore.GREEN}{content}(Waiting for input. Your input can only be text or image path each time, you can press Enter once to input multiple times. Press Enter twice to finish the entire input.):{Style.RESET_ALL}")
user_input_lines = []
except Exception as e:
logging.debug(f"Consumer group may already exist: {e}")


while True:
try:
status = self._interactor.get_workflow(workflow_id=workflow_instance_id).status #获取执行状态
if status in terminal_status:
break
data_flag = False
content = None
# logging.info(f"Checking workflow status: {workflow_instance_id}")
workflow_status = workflow_client.get_workflow_status(workflow_instance_id)
if workflow_status.status not in running_status:
logging.info(f"Workflow {workflow_instance_id} is not running, exiting...")
break

# read new messages from consumer group
messages = container.get_connector('redis_stream_client')._client.xreadgroup(
group_name, consumer_name, {stream_name: ">"}, count=1
)
# Convert byte data to string
messages = [
(stream, [(message_id, {k.decode('utf-8'): v.decode('utf-8') for k, v in message.items()}) for message_id, message in message_list])
for stream, message_list in messages
]
# logging.info(f"Messages: {messages}")

for stream, message_list in messages:
for message_id, message in message_list:
data_flag, content = self.process_message(message)
# confirm message has been processed
container.get_connector('redis_stream_client')._client.xack(
stream_name, group_name, message_id
)
if data_flag:
contents = []
while True:
line = input(f"{Fore.GREEN}>>>{Style.RESET_ALL}")
if line == "":
break
user_input_lines.append(line)
logging.info(f"User input lines: {user_input_lines}")
print(f"{Fore.GREEN}{content}(Waiting for input. Your input can only be text or image path each time, you can press Enter once to input multiple times. Press Enter twice to finish the entire input.):{Style.RESET_ALL}")
user_input_lines = []
while True:
line = input(f"{Fore.GREEN}>>>{Style.RESET_ALL}")
if line == "":
break
user_input_lines.append(line)
logging.info(f"User input lines: {user_input_lines}")



for user_input in user_input_lines:
if self.is_url(user_input) or self.is_file(user_input):
contents.append({
"type": "image_url",
"data": user_input
})
else:
contents.append({
"type": "text",
"data": user_input
})
if len(contents) > 0:
break
result = {
"agent_id": workflow_instance_id,
"messages": [
{
"role": "user",
"content": contents
}
],
"kwargs": {}
}
container.get_connector('redis_stream_client')._client.xadd(f"{workflow_instance_id}_input", {"payload":json.dumps(result, ensure_ascii=False) })
# Sleep for the specified interval before checking for new messages again
# logging.info(f"Sleeping for {poll_interval} seconds, waiting for {stream_name} ...")
sleep(poll_interval)
except Exception as e:
logging.error(f"Error while listening to stream: {e}")
sleep(poll_interval) # Wait before retrying
self.stop_interactor()
for user_input in user_input_lines:
if self.is_url(user_input) or self.is_file(user_input):
contents.append({
"type": "image_url",
"data": user_input
})
else:
contents.append({
"type": "text",
"data": user_input
})
if len(contents) > 0:
break
result = {
"agent_id": workflow_instance_id,
"messages": [
{
"role": "user",
"content": contents
}
],
"kwargs": {}
}
container.get_connector('redis_stream_client')._client.xadd(f"{workflow_instance_id}_input", {"payload":json.dumps(result, ensure_ascii=False) })
# Sleep for the specified interval before checking for new messages again
# logging.info(f"Sleeping for {poll_interval} seconds, waiting for {stream_name} ...")
sleep(poll_interval)
except Exception as e:
logging.error(f"Error while listening to stream: {e}")
sleep(poll_interval) # Wait before retrying
self.stop_interactor()
except KeyboardInterrupt:
logging.info("\nDetected Ctrl+C, stopping workflow...")
if workflow_instance_id is not None:
self._interactor._executor.terminate(workflow_id=workflow_instance_id)
raise

def stop_interactor(self):
self._task_handler_interactor.stop_processes()

def start_processor(self):
worker_config = build_from_file(self._config_path)
self._task_handler_processor = TaskHandler(worker_config=worker_config, workers=self._workers)
self._task_handler_processor.start_processes()
workflow_instance_id = self._processor.start_workflow_with_input(workflow_input={})
user_input = input(f"{Fore.GREEN}Please input a folder path of images:(WaitPress Enter to finish the entire input.):\n>>>{Style.RESET_ALL}")

image_items = []
idx = 0
for image_file in Path(user_input).iterdir():
if image_file.is_file() and image_file.suffix in ['.png', '.jpg', '.jpeg']:
image_items.append({
"type": "image_url",
"resource_id": str(idx),
"data": str(image_file)
})
idx += 1
result = {
"content": image_items
}
container.get_connector('redis_stream_client')._client.xadd(f"image_process", {"payload":json.dumps(result, ensure_ascii=False) })
while True:
status = self._processor.get_workflow(workflow_id=workflow_instance_id).status
if status == 'COMPLETED':
break

sleep(1)
self.stop_processor()

workflow_instance_id = None
try:
worker_config = build_from_file(self._config_path)
self._task_handler_processor = TaskHandler(worker_config=worker_config, workers=self._workers)
self._task_handler_processor.start_processes()
workflow_instance_id = self._processor.start_workflow_with_input(workflow_input={})
user_input = input(f"{Fore.GREEN}Please input a folder path of images:(WaitPress Enter to finish the entire input.):\n>>>{Style.RESET_ALL}")

image_items = []
idx = 0
for image_file in Path(user_input).iterdir():
if image_file.is_file() and image_file.suffix in ['.png', '.jpg', '.jpeg']:
image_items.append({
"type": "image_url",
"resource_id": str(idx),
"data": str(image_file)
})
idx += 1
result = {
"content": image_items
}
container.get_connector('redis_stream_client')._client.xadd(f"image_process", {"payload":json.dumps(result, ensure_ascii=False) })
while True:
status = self._processor.get_workflow(workflow_id=workflow_instance_id).status
if status in terminal_status:
break

sleep(1)
self.stop_processor()
except KeyboardInterrupt:
logging.info("\nDetected Ctrl+C, stopping workflow...")
if workflow_instance_id is not None:
self._processor._executor.terminate(workflow_id=workflow_instance_id)
raise

def stop_processor(self):
self._task_handler_processor.stop_processes()
Expand Down
Loading

0 comments on commit 95b0a01

Please sign in to comment.