Spaces:
Sleeping
Sleeping
File size: 2,751 Bytes
3a235a9 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 |
import asyncio
import time
from dataclasses import dataclass, field
from time import sleep
from rich.table import Table
from aworld.output.utils import consume_content
from rich.status import Status
from rich.console import Console
from aworld.output import MessageOutput, WorkSpace
from aworld.output.base import StepOutput, ToolResultOutput
from aworld.output.ui.base import AworldUI
import json
@dataclass
class StreamAworldUI(AworldUI):
console: Console = field(default_factory=Console)
status: Status = None
workspace: WorkSpace = None
gaia_output_line_tag = "GA_FMT_CONTENT:"
async def message_output(self, __output__: MessageOutput):
result = []
async def __log_item(item):
result.append(item)
self.console.print(f"{self.gaia_output_line_tag} {json.dumps(item)}", end="")
if __output__.reason_generator or __output__.response_generator:
if __output__.reason_generator:
await consume_content(__output__.reason_generator, __log_item)
if __output__.reason_generator:
await consume_content(__output__.response_generator, __log_item)
else:
await consume_content(__output__.reasoning, __log_item)
await consume_content(__output__.response, __log_item)
# if __output__.tool_calls:
# await consume_content(__output__.tool_calls, __log_item)
self.console.print("")
async def tool_result(self, output: ToolResultOutput):
"""
tool_result
"""
table = Table(show_header=False, header_style="bold magenta", title=f"Call Tools#ID_{output.origin_tool_call.id}")
table.add_column("name", style="dim", width=12)
table.add_column("content")
table.add_row("function_name", output.origin_tool_call.function.name)
table.add_row("arguments", output.origin_tool_call.function.arguments)
table.add_row("result", output.data)
self.console.print(table)
async def step(self, output: StepOutput):
if output.status == "START":
self.console.print(f"[bold green]{output.name} ✈️START ...")
self.status = self.console.status(f"[bold green]{output.name} RUNNING ...")
self.status.start()
elif output.status == "FINISHED":
self.status.stop()
self.console.print(f"[bold green]{output.name} 🛬FINISHED ...")
elif output.status == "FAILED":
self.status.stop()
self.console.print(f"[bold red]{output.name} 💥FAILED ...")
else:
self.status.stop()
self.console.print(f"============={output.name} ❓❓❓UNKNOWN#{output.status} ======================")
|