Upload 9 files

aworld/output/README.md

@@ -0,0 +1,156 @@
+# AWorld Output Module
+
+The Output module is a flexible and extensible system for managing outputs and artifacts in the AWorld framework.
+
+## Key Features
+
+- **Unified Output Management**: Centralized management of all output types (messages, artifacts, tool results, etc.) through a flexible Outputs interface.
+- **Support for Multiple Output Types**: Handles text, code, files, tool calls, and custom outputs, enabling rich interaction and extensibility.
+- **Async & Sync Streaming**: Provides both asynchronous and synchronous streaming of outputs, supporting real-time and batch processing scenarios.
+- **Output Aggregation & Dispatch**: Aggregates outputs from various sources and dispatches them to different consumers or UIs.
+- **Extensible Output Channels**: Easily extendable output channels and renderers for custom UI or integration needs.
+- **Integration with Task & Workspace**: Seamlessly integrates with Task and WorkSpace modules for collaborative, versioned, and observable output management.
+- **Real-time, Batch, and Streaming Modes**: Supports real-time, batch, and streaming output modes for flexible workflow requirements.
+- **Observer Pattern Support**: Built-in observer pattern for real-time updates and notifications on output or artifact changes.
+- **Easy Customization**: Designed for easy extension and customization to fit various application scenarios.
+- **Rich UI Rendering**: Supports diverse output rendering with pluggable UI renderers for CLI, web, and custom frontends.
+- **Decoupled UI & Output Types**: UI rendering is decoupled from output types, enabling flexible presentation and interaction.
+- **Real-time Interactive Display**: Enables real-time, streaming, and interactive output display in various UI environments.
+- **UI Extensibility**: Easy to extend and customize UI components to fit different user experiences and workflows.
+
+## Class Diagram
+
+```mermaid
+classDiagram
+    direction TB
+    %% Output Related Classes
+    class Output {
+        +metadata: Dict
+        +parts: List[OutputPart]
+    }
+    class OutputPart {
+        +content: Any
+        +metadata: Dict
+    }
+    Output *-- OutputPart
+    class Outputs {
+        <<abstract>>
+        +add_output(output: Output)
+        +sync_add_output(output: Output)
+        +stream_events()
+        +sync_stream_events()
+        +mark_completed()
+    }
+    class AsyncOutputs {
+        +add_output(output: Output)
+        +sync_add_output(output: Output)
+        +stream_events()
+        +sync_stream_events()
+    }
+    class DefaultOutputs {
+        +_outputs: List[Output]
+        +add_output(output: Output)
+        +sync_add_output(output: Output)
+        +stream_events()
+        +sync_stream_events()
+        +mark_completed()
+    }
+    class StreamingOutputs {
+        +input: Any
+        +usage: dict
+        +is_complete: bool
+        +_output_queue: asyncio.Queue[Output]
+        +_visited_outputs: List[Output]
+        +_stored_exception: Exception
+        +_run_impl_task: asyncio.Task
+        +add_output(output: Output)
+        +stream_events()
+        +mark_completed()
+    }
+    class MessageOutput {
+        +source: Any
+        +reason_generator: Any
+        +response_generator: Any
+        +reasoning: str
+        +response: Any
+        +has_reasoning: bool
+        +finished: bool
+    }
+    class Artifact {
+        +artifact_id: str
+        +artifact_type: ArtifactType
+        +content: Any
+        +metadata: Dict
+    }
+    class ToolCallOutput {
+        +tool_id: str
+        +content: Any
+        +metadata: Dict
+    }
+    class ToolResultOutput {
+        +tool_id: str
+        +params: Any
+        +content: Any
+        +metadata: Dict
+    }
+    class SearchOutput {
+        +artifact_id: str
+        +artifact_type: ArtifactType
+        +content: Any
+        +metadata: Dict
+    }
+
+    %% Inheritance Relationships
+    Outputs <|-- AsyncOutputs
+    Outputs <|-- DefaultOutputs
+    AsyncOutputs <|-- StreamingOutputs
+    Output <|-- MessageOutput
+    Output <|-- Artifact
+    Output <|-- ToolCallOutput
+    Output <|-- ToolResultOutput
+    ToolResultOutput <|-- SearchOutput
+
+    %% Aggregation/Composition
+    Outputs o-- Output
+    DefaultOutputs o-- Output
+    StreamingOutputs o-- Output
+
+    %% Workspace Related Classes
+    class WorkSpace {
+        +workspace_id: str
+        +name: str
+        +created_at: str
+        +updated_at: str
+        +metadata: Dict
+        +artifacts: List[Artifact]
+        +observers: List[WorkspaceObserver]
+        +repository: ArtifactRepository
+        +create_artifact()
+        +add_artifact()
+        +get_artifact()
+        +update_artifact()
+        +delete_artifact()
+        +list_artifacts()
+        +add_observer()
+        +remove_observer()
+        +save()
+        +load()
+    }
+    class WorkspaceObserver
+    class ArtifactRepository
+
+    WorkSpace o-- Artifact
+    WorkSpace o-- WorkspaceObserver
+    WorkSpace o-- ArtifactRepository
+
+    %% Task Related Classes
+    class Task {
+        +name: str
+        +input: Any
+        +outputs: Outputs
+    }
+
+    class Outputs
+
+    Task o-- Outputs
+```

aworld/output/__init__.py

@@ -0,0 +1,31 @@
+from aworld.output.base import Output, SearchOutput, SearchItem, ToolResultOutput, MessageOutput, ToolCallOutput, \
+    RUN_FINISHED_SIGNAL
+from aworld.output.artifact import Artifact, ArtifactType
+from aworld.output.code_artifact import CodeArtifact, ShellArtifact
+from aworld.output.outputs import Outputs, StreamingOutputs
+from aworld.output.workspace import WorkSpace
+from aworld.output.observer import WorkspaceObserver,get_observer
+from aworld.output.storage.artifact_repository import ArtifactRepository, LocalArtifactRepository
+from aworld.output.ui.base import AworldUI,PrinterAworldUI
+__all__ = [
+    "Output",
+    "Artifact",
+    "ArtifactType",
+    "CodeArtifact",
+    "ShellArtifact",
+    "WorkSpace",
+    "ArtifactRepository",
+    "LocalArtifactRepository",
+    "WorkspaceObserver",
+    "get_observer",
+    "SearchOutput",
+    "SearchItem",
+    "MessageOutput",
+    "ToolCallOutput",
+    "ToolResultOutput",
+    "Outputs",
+    "StreamingOutputs",
+    "RUN_FINISHED_SIGNAL",
+    "AworldUI",
+    "PrinterAworldUI"
+]

aworld/output/artifact.py

@@ -0,0 +1,155 @@
+import uuid
+from datetime import datetime
+from enum import Enum, auto
+from typing import Dict, Any, Optional, ClassVar
+from pydantic import Field, field_validator, model_validator, BaseModel
+
+from aworld.output.base import Output
+
+
+class ArtifactType(Enum):
+    """Defines supported artifact types"""
+    TEXT = "TEXT"
+    CODE = "CODE"
+    MARKDOWN = "MARKDOWN"
+    HTML = "HTML"
+    SVG = "SVG"
+    IMAGE = "IMAGE"
+    JSON = "JSON"
+    CSV = "CSV"
+    TABLE = "TABLE"
+    CHART = "CHART"
+    DIAGRAM = "DIAGRAM"
+    MCP_CALL = "MCP_CALL"
+    TOOL_CALL = "TOOL_CALL"
+    LLM_OUTPUT = "LLM_OUTPUT"
+    WEB_PAGES = "WEB_PAGES"
+    DIR = "DIR"
+    CUSTOM = "CUSTOM"
+
+
+
+class ArtifactStatus(Enum):
+    """Artifact status"""
+    DRAFT = auto()      # Draft status
+    COMPLETE = auto()   # Completed status
+    EDITED = auto()     # Edited status
+    ARCHIVED = auto()   # Archived status
+
+class ArtifactAttachment(BaseModel):
+    filename: str = Field(..., description="Filename")
+    content: str = Field(..., description="Content", exclude=True)
+    mime_type: str = Field(..., description="MIME type")
+
+
+class Artifact(Output):
+    """
+    Represents a specific content generation result (artifact)
+    
+    Artifacts are the basic units of Artifacts technology, representing a structured content unit
+    Can be code, markdown, charts, and various other formats
+    """
+
+    artifact_id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Unique identifier for the artifact")
+    artifact_type: ArtifactType = Field(..., description="Type of the artifact")
+    content: Any = Field(..., description="Content of the artifact")
+    metadata: Dict[str, Any] = Field(default_factory=dict, description="Metadata associated with the artifact")
+    created_at: str = Field(default_factory=lambda: datetime.now().isoformat(), description="Creation timestamp")
+    updated_at: str = Field(default_factory=lambda: datetime.now().isoformat(), description="Last updated timestamp")
+    status: ArtifactStatus = Field(default=ArtifactStatus.COMPLETE, description="Current status of the artifact")
+    current_version: str = Field(default="", description="Current version of the artifact")
+    version_history: list = Field(default_factory=list, description="History of versions for the artifact")
+    create_file: bool = Field(default=False, description="Flag to indicate if a file should be created")
+    attachments: Optional[list[ArtifactAttachment]] = Field(default_factory=list, description="Attachments associated with the artifact")
+
+    def _record_version(self, description: str) -> None:
+        """Record current state as a new version"""
+        version = {
+            "timestamp": datetime.now().isoformat(),
+            "description": description,
+            "status": self.status
+        }
+        self.version_history.append(version)
+        self.updated_at = version["timestamp"]
+
+    def update_content(self, content: Any, description: str = "Content update") -> None:
+        """
+        Update artifact content and record version
+        
+        Args:
+            content: New content
+            description: Update description
+        """
+        self.content = content
+        self.status = ArtifactStatus.EDITED
+        self._record_version(description)
+
+    def update_metadata(self, metadata: Dict[str, Any]) -> None:
+        """
+        Update artifact metadata
+        
+        Args:
+            metadata: New metadata (will be merged with existing metadata)
+        """
+        self.metadata.update(metadata)
+        self.updated_at = datetime.now().isoformat()
+
+    def mark_complete(self) -> None:
+        """Mark the artifact as complete"""
+        self.status = ArtifactStatus.COMPLETE
+        self.updated_at = datetime.now().isoformat()
+        self._record_version("Marked as complete")
+
+    def archive(self) -> None:
+        """Archive the artifact"""
+        self.status = ArtifactStatus.ARCHIVED
+        self._record_version("Artifact archived")
+
+    def get_version(self, index: int) -> Optional[Dict[str, Any]]:
+        """Get version at the specified index"""
+        if 0 <= index < len(self.version_history):
+            return self.version_history[index]
+        return None
+
+    def revert_to_version(self, index: int) -> bool:
+        """Revert to a specific version"""
+        version = self.get_version(index)
+        if version:
+            self.content = version["content"]
+            self.status = version["status"]
+            self._record_version(f"Reverted to version {index}")
+            return True
+        return False
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert artifact to dictionary"""
+        return {
+            "artifact_id": self.artifact_id,
+            "artifact_type": self.artifact_type.value,
+            "content": self.content,
+            "metadata": self.metadata,
+            "created_at": self.created_at,
+            "updated_at": self.updated_at,
+            "status": self.status.name,
+            "version_count": len(self.version_history)
+        }
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "Artifact":
+        """Create an artifact instance from a dictionary"""
+        artifact_type = ArtifactType(data["artifact_type"])
+        artifact = cls(
+            artifact_type=artifact_type,
+            content=data["content"],
+            metadata=data["metadata"],
+            artifact_id=data.get("artifact_id", str(uuid.uuid4()))
+        )
+        artifact.created_at = data["created_at"]
+        artifact.updated_at = data["updated_at"]
+        artifact.status = ArtifactStatus[data["status"]]
+
+        # If version history exists, restore it as well
+        if "version_history" in data:
+            artifact.version_history = data["version_history"]
+
+        return artifact

aworld/output/base.py

@@ -0,0 +1,405 @@
+import json
+import logging
+from builtins import anext
+from datetime import datetime
+from typing import Any, Dict, Generator, AsyncGenerator, Optional
+
+from pydantic import Field, BaseModel, model_validator
+
+from aworld.models.model_response import ModelResponse, ToolCall
+
+
+class OutputPart(BaseModel):
+    content: Any
+    metadata: Optional[Dict[str, Any]] = Field(default_factory=dict, description="metadata")
+
+    @model_validator(mode='after')
+    def setup_metadata(self):
+        # Ensure metadata is initialized
+        if self.metadata is None:
+            self.metadata = {}
+        return self
+    
+
+class Output(BaseModel):
+    metadata: Optional[Dict[str, Any]] = Field(default_factory=dict, description="metadata")
+    parts: Any = Field(default_factory=list, exclude=True, description="parts of Output")
+    data: Any = Field(default=None, exclude=True, description="Output Data")
+
+    @model_validator(mode='after')
+    def setup_defaults(self):
+        # Ensure metadata and parts are initialized
+        if self.metadata is None:
+            self.metadata = {}
+        if self.parts is None:
+            self.parts = []
+        return self
+
+    def add_part(self, content: Any):
+        if self.parts is None:
+            self.parts = []
+        self.parts.append(OutputPart(content=content))
+
+    def output_type(self):
+        return "default"
+
+
+class ToolCallOutput(Output):
+
+    @classmethod
+    def from_tool_call(cls, tool_call: ToolCall):
+        return cls(data = tool_call)
+
+    def output_type(self):
+        return "tool_call"
+
+class ToolResultOutput(Output):
+
+    origin_tool_call: Optional[ToolCall] = Field(default=None, description="origin tool call", exclude=True)
+
+    image: str = Field(default=None)
+
+    images: list[str] = Field(default_factory=list)
+
+    tool_type: str = Field(default=None)
+
+    tool_name: str = Field(default=None)
+
+    def output_type(self):
+        return "tool_call_result"
+
+    pass
+
+class RunFinishedSignal(Output):
+
+    def output_type(self):
+        return "finished_signal"
+
+RUN_FINISHED_SIGNAL = RunFinishedSignal()
+
+
+class MessageOutput(Output):
+
+    """
+    MessageOutput structure of LLM output
+    if you want to get the only response, you must first call reasoning_generator or set parameter only_response to True , then call response_generator
+    if you model not reasoning, you do not need care about reasoning_generator and reasoning
+
+    1. source: async/sync generator of the message
+    2. reasoning_generator: async/sync reasoning generator of the message
+    3. response_generator: async/sync response generator of the message;
+    4. reasoning: reasoning of the message
+    5. response: response of the message
+    6. tool_calls
+
+    """
+
+    source: Any = Field(default=None, exclude=True, description="Source of the message")
+    
+    reason_generator: Any = Field(default=None, exclude=True, description="reasoning generator of the message")
+    response_generator: Any = Field(default=None, exclude=True, description="response generator of the message")
+
+    """
+    result
+    """
+    reasoning: str = Field(default=None, description="reasoning of the message")
+    response: Any = Field(default=None, description="response of the message")
+    tool_calls: list[ToolCallOutput] = Field(default_factory=list, description="tool_calls")
+
+
+    """
+    other config
+    """
+    reasoning_format_start: str = Field(default="<think>", description="reasoning format start of the message")
+    reasoning_format_end: str = Field(default="</think>", description="reasoning format end of the message")
+
+    json_parse: bool = Field(default=False, description="json parse of the message", exclude=True)
+    has_reasoning: bool = Field(default=False, description="has reasoning of the message")
+    finished: bool = Field(default=False, description="finished of the message")
+
+    @model_validator(mode='after')
+    def setup_generators(self):
+        """
+        Setup generators for reasoning and response
+        """
+        source = self.source
+
+        # if ModelResponse
+        if isinstance(self.source, ModelResponse):
+            source = self.source.content
+            if self.source.tool_calls:
+                [self.tool_calls.append(ToolCallOutput.from_tool_call(tool_call)) for tool_call in
+                 self.source.tool_calls]
+
+        if source is not None and isinstance(source, AsyncGenerator):
+            # Create empty generators first, they will be initialized when actually used
+            self.reason_generator = self.__aget_reasoning_generator()
+            self.response_generator = self.__aget_response_generator()
+        elif source is not None and isinstance(source, Generator):
+            self.reason_generator, self.response_generator = self.__split_reasoning_and_response__()
+        elif source is not None and isinstance(source, str):
+            self.reasoning, self.response = self.__resolve_think__(source)
+        return self
+
+    async def get_finished_reasoning(self):
+        if self.reasoning:
+            return self.reasoning
+        else:
+            if self.has_reasoning and not self.reasoning:
+                async for reason in self.reason_generator:
+                    pass
+                return self.reasoning
+            else:
+                return self.reasoning
+
+    async def get_finished_response(self):
+        if self.response:
+            return self.response
+        else:
+            if self.response_generator:
+                async for item in self.response_generator:
+                    pass
+            return self.response
+    
+    async def __aget_reasoning_generator(self) -> AsyncGenerator[str, None]:
+        """
+        Get reasoning content as async generator
+        """
+        if not self.has_reasoning:
+            yield ""
+            self.reasoning = ""
+            return  
+        
+        reasoning_buffer = ""
+        is_in_reasoning = False
+        if self.reasoning and len(self.reasoning) > 0:
+            yield self.reasoning
+            return
+        
+        try:
+            while True:
+                chunk = await anext(self.source)
+                chunk_content = self.get_chunk_content(chunk)
+                if not chunk_content:
+                    continue
+                if chunk_content.startswith(self.reasoning_format_start):
+                    is_in_reasoning = True
+                    reasoning_buffer = chunk_content
+                    yield chunk_content
+                elif chunk_content.endswith(self.reasoning_format_end) and is_in_reasoning:
+                    reasoning_buffer += chunk_content
+                    yield chunk_content
+                    self.reasoning = reasoning_buffer
+                    break
+                elif is_in_reasoning:
+                    reasoning_buffer += chunk_content
+                    yield chunk_content
+        except StopAsyncIteration:
+            logging.info("StopAsyncIteration")
+
+    async def __aget_response_generator(self) -> AsyncGenerator[str, None]:
+        """
+        Get response content as async generator
+
+        if has_reasoning is True, system will first call reasoning_generator if you not call it;
+        else it will return content contains reasoning and response
+        """
+        response_buffer = ""
+
+        if self.response and len(self.response) > 0:
+            yield self.response
+            return
+        
+        # if has_reasoning is True, system will first call reasoning_generator if you not call it;
+        if self.has_reasoning and not self.reasoning:
+            async for reason in self.reason_generator:
+                pass
+
+        try:
+            while True:
+                chunk = await anext(self.source)
+                chunk_content = self.get_chunk_content(chunk)
+
+                if not chunk_content:
+                    continue
+                response_buffer += chunk_content
+                yield chunk_content
+        except StopAsyncIteration:
+            self.finished = True
+            self.response = self.__resolve_json__(response_buffer, self.json_parse)
+
+    def get_chunk_content(self, chunk):
+        if isinstance(chunk, ModelResponse):
+            return chunk.content
+        else:
+            return chunk
+
+    def __split_reasoning_and_response__(self) -> tuple[Generator[str, None, None], Generator[str, None, None]]: # type: ignore
+        """
+        Split source into reasoning and response generators for sync source
+        Returns:
+            tuple: (reasoning_generator, response_generator)
+        """
+        if not self.has_reasoning:
+            yield ""
+            self.reasoning = ""
+            return  
+        
+        if not isinstance(self.source, Generator):
+            raise ValueError("Source must be a Generator")
+
+        def reasoning_generator():
+            if self.reasoning and len(self.reasoning) > 0:
+                yield self.reasoning
+                return
+
+            reasoning_buffer = ""
+            is_in_reasoning = False
+            
+            try:
+                while True:
+                    chunk = next(self.source)
+                    chunk_content = self.get_chunk_content(chunk)
+                    if chunk_content.startswith(self.reasoning_format_start):
+                        is_in_reasoning = True
+                        reasoning_buffer = chunk_content
+                        yield chunk_content
+                    elif chunk_content.endswith(self.reasoning_format_end) and is_in_reasoning:
+                        reasoning_buffer += chunk_content
+                        self.reasoning = reasoning_buffer
+                        yield chunk_content
+                        break
+                    elif is_in_reasoning:
+                        yield chunk_content
+                        reasoning_buffer += chunk_content
+            except StopIteration:
+                print("StopIteration")
+                self.reasoning = reasoning_buffer
+
+        def response_generator():
+            if self.response and len(self.response) > 0:
+                yield self.response
+                return
+            
+            # if has_reasoning is True, system will first call reasoning_generator if you not call it;
+            if self.has_reasoning and not self.reasoning:
+                for reason in self.reason_generator:
+                    pass
+            
+            response_buffer = ""
+            try:
+                while True:
+                    chunk = next(self.source)
+                    chunk_content = self.get_chunk_content(chunk)
+                    response_buffer += chunk_content
+                    self.response = response_buffer
+                    yield chunk_content
+            except StopIteration:
+                self.response = self.__resolve_json__(response_buffer,self.json_parse)
+                self.finished = True
+
+
+        return reasoning_generator(), response_generator()
+
+    def __resolve_think__(self, content):
+        import re
+        start_tag = self.reasoning_format_start.replace("<", "").replace(">", "")
+        end_tag = self.reasoning_format_end.replace("<", "").replace(">", "")
+
+        llm_think = ""
+        match = re.search(
+            rf"<{re.escape(start_tag)}(.*?)>(.|\n)*?<{re.escape(end_tag)}>",
+            content,
+            flags=re.DOTALL,
+        )
+        if match:
+            llm_think = match.group(0).replace("<think>", "").replace("</think>", "")
+        llm_result = re.sub(
+            rf"<{re.escape(start_tag)}(.*?)>(.|\n)*?<{re.escape(end_tag)}>",
+            "",
+            content,
+            flags=re.DOTALL,
+        )
+        llm_result = self.__resolve_json__(llm_result, self.json_parse)
+
+        return llm_think, llm_result
+
+    def __resolve_json__(self, content, json_parse = False):
+        if json_parse:
+            if content.__contains__("```json"):
+                content = content.replace("```json", "").replace("```", "")
+            return json.loads(content)
+        return content
+
+    def output_type(self):
+        return "message_output"
+
+class StepOutput(Output):
+    name: str
+    step_num: int
+    alias_name: Optional[str] = Field(default=None, description="alias_name of the step")
+    status: Optional[str] = Field(default="START", description="step_status")
+    started_at: str = Field(default_factory=lambda: datetime.now().isoformat(), description="started at")
+    finished_at: str = Field(default_factory=lambda: datetime.now().isoformat(), description="finished at")
+
+    @classmethod
+    def build_start_output(cls, name, step_num, alias_name=None, data=None):
+        return cls(name=name, step_num=step_num, alias_name=alias_name, data=data)
+
+    @classmethod
+    def build_finished_output(cls, name, step_num, alias_name=None, data=None):
+        return cls(name=name, step_num=step_num, alias_name=alias_name, status='FINISHED', data=data)
+
+    @classmethod
+    def build_failed_output(cls, name, step_num, alias_name=None, data=None):
+        return cls(name=name, step_num=step_num, alias_name=alias_name, status='FAILED', data=data)
+
+    def output_type(self):
+        return "step_output"
+
+    @property
+    def show_name(self):
+        return self.alias_name if self.alias_name else self.name
+
+
+
+class SearchItem(BaseModel):
+    title: str = Field(default="", description="search result title")
+    url: str = Field(default="", description="search result url")
+    snippet: str = Field(default="", description="search result snippet")
+    content: str = Field(default="", description="search content", exclude=True)
+    raw_content: Optional[str] = Field(default="", description="search raw content", exclude=True)
+    metadata: Optional[Dict[str, Any]] = Field(default_factory=dict, description="metadata")
+
+
+class SearchOutput(ToolResultOutput):
+    query: str = Field(..., description="Search query string")
+    results: list[SearchItem] = Field(default_factory=list, description="List of search results")
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "SearchOutput":
+        if not isinstance(data, dict):
+            data = {}
+
+        query = data.get("query")
+        if query is None:
+            raise ValueError("query is required")
+
+        results_data = data.get("results", [])
+        
+        search_items = []
+        for result in results_data:
+            if isinstance(result, SearchItem):
+                search_items.append(result)
+            elif isinstance(result, dict):
+                search_items.append(SearchItem(**result))
+            else:
+                raise ValueError(f"Invalid result type: {type(result)}")
+
+        return cls(
+            query=query,
+            results=search_items
+        )
+
+    def output_type(self):
+        return "search_output"

aworld/output/code_artifact.py

@@ -0,0 +1,277 @@
+import uuid
+from typing import Any, Optional, Dict, List
+
+from pydantic import Field
+
+from aworld.output.artifact import Artifact, ArtifactType, ArtifactAttachment
+
+CODE_FILE_EXTENSION_MAP = {
+                "python": "py",
+                "java": "java",
+                "javascript": "js",
+                "typescript": "ts",
+                "html": "html",
+                "css": "css",
+                "c": "c",
+                "cpp": "cpp",
+                "csharp": "cs",
+                "go": "go",
+                "rust": "rs",
+                "ruby": "rb",
+                "php": "php",
+                "swift": "swift",
+                "kotlin": "kt",
+                "scala": "scala",
+                "markdown": "md",
+                "txt": "txt",
+                "shell": "sh",
+                "bash": "sh",
+                "sh": "sh",
+                "zsh": "zsh",
+                "powershell": "ps1",
+                "cmd": "cmd",
+                "bat": "bat"
+            }
+
+
+class CodeArtifact(Artifact):
+    code_interceptor: Any = Field(default=None, description="code executor type")
+
+    def __init__(self, artifact_type: ArtifactType, content: Any, code_type: Optional[str], code_version: Optional[str],
+                 code_interceptor_provider: Optional[str] = None,
+                 artifact_id: Optional[str] = None, render_type: Optional[str] = None, **kwargs):
+        # Extract filename from the first line of the content
+        filename = self.extract_filename(content)
+        
+        # Initialize metadata, including any passed in kwargs
+        metadata = {
+            "code_type": code_type,
+            "code_version": code_version,
+            "code_interceptor_provider": code_interceptor_provider,
+            "filename": filename  # Store filename in metadata
+        }
+        
+        # Merge additional metadata from kwargs if provided
+        if 'metadata' in kwargs:
+            metadata.update(kwargs['metadata'])
+            del kwargs['metadata']  # Remove metadata from kwargs to avoid multiple values
+
+        super().__init__(
+            artifact_type=artifact_type,
+            content=content,
+            metadata=metadata,
+            artifact_id=artifact_id,
+            render_type=render_type,
+            **kwargs
+        )
+        self.archive()
+        self.code_interceptor = self.init_code_interceptor(code_interceptor_provider)
+
+    @staticmethod
+    def extract_filename(content: Any) -> Optional[str]:
+        """Extract filename from the first line of the code block comment."""
+        if isinstance(content, str):
+            lines = content.splitlines()
+            if lines:
+                first_line = lines[0].strip()
+                # Check if the first line is a shebang for bash or other interpreters
+                if first_line in ["# /bin/bash", "#!/bin/bash", "#!/usr/bin/env bash", 
+                                   "#!/bin/sh", "#!/usr/bin/env python", 
+                                   "#!/usr/bin/env python3"]:
+                    return None  # Do not return a filename
+                # Check for common comment styles in various languages
+                if first_line.startswith("#"):  # Python, Ruby, Shell
+                    return first_line[1:].strip()  # Remove the comment symbol
+                elif first_line.startswith("//"):  # Java, JavaScript, C, C++
+                    return first_line[2:].strip()  # Remove the comment symbol
+                elif first_line.startswith("/*") and "*/" in first_line:  # C, C++
+                    return first_line.split("*/")[0][2:].strip()  # Remove comment symbols
+                elif first_line.startswith("<!--"):  # HTML
+                    return first_line[4:].strip()  # Remove the comment symbol
+                # Add more languages as needed
+        return None  # Return None if filename is unknown
+
+    @classmethod
+    def build_artifact(cls,
+                       content: Any,
+                       code_type: Optional[str] = None,
+                       code_version: Optional[str] = None,
+                       code_interceptor_provider: Optional[str] = None,
+                       artifact_id: Optional[str] = None,
+                       render_type: Optional[str] = None,
+                       **kwargs) -> "CodeArtifact":
+
+        # Create CodeArtifact instance
+        if code_type in ['shell', 'sh', 'bash', 'zsh']:
+            return ShellArtifact(
+                artifact_type=ArtifactType.CODE,
+                content=content,
+                code_version=code_version,
+                code_interceptor_provider=code_interceptor_provider,
+                artifact_id=artifact_id,
+                render_type=render_type,
+                **kwargs
+            )
+        elif code_type in ['html']:
+            return HtmlArtifact(
+                content=content,
+                artifact_id=artifact_id,
+                **kwargs
+            )
+
+        return cls(
+            artifact_type=ArtifactType.CODE,
+            content=content,
+            code_type=code_type,
+            code_version=code_version,
+            code_interceptor_provider=code_interceptor_provider,
+            artifact_id=artifact_id,
+            render_type=render_type,
+            **kwargs
+        )
+
+    @classmethod
+    def from_code_content(cls, artifact_type: ArtifactType,
+                          content: Any,
+                          render_type: Optional[str] = None,
+                          **kwargs) -> List["CodeArtifact"]:
+        code_blocks = cls.extract_model_output_to_code_content(content)  # Extract code blocks
+        artifacts = []  # List to store CodeArtifact instances
+
+        for block in code_blocks:
+            code_type = block['language']
+            code_version = "1.0"
+
+            if code_type in ['python', 'javascript', 'java']:
+                code_interceptor_provider = "default_interceptor"
+            elif code_type in ['shell', 'sh', 'bash', 'zsh']:
+                code_interceptor_provider = "shell_interceptor"
+            else:
+                code_interceptor_provider = "generic_interceptor"
+
+            artifact = cls.create_artifact(
+                artifact_type=ArtifactType.CODE,
+                content=block['content'],
+                code_type=code_type,
+                code_version=code_version,
+                code_interceptor_provider=code_interceptor_provider,
+                artifact_id=block['artifact_id'],  # Use extracted artifact_id
+                render_type=render_type,
+                **kwargs
+            )
+            artifacts.append(artifact)  # Add to the list
+
+        return artifacts  # Return the list of CodeArtifact instances
+
+    def init_code_interceptor(self, code_interceptor_provider):
+        pass
+
+    @classmethod
+    def extract_model_output_to_code_content(cls, content):
+        """
+        Extract code blocks from markdown content using mistune.
+        
+        First extracts all code blocks enclosed in triple backticks,
+        then determines the language for each block.
+        """
+
+        try:
+            import mistune
+        except ImportError:
+            # install mistune
+            import subprocess
+            subprocess.run(["pip", "install", "mistune>=3.0.0"], check=True)
+            import mistune
+
+        code_blocks = []
+        
+        #
+        extracted_blocks = []
+        
+        # create custom Render
+        class CustomRenderer(mistune.HTMLRenderer):
+            def block_code(self, code, info=None):
+                language = info.split()[0] if info else 'unknown'
+                extracted_blocks.append({
+                    "content": code,
+                    "language": language
+                })
+                return ""
+
+        # create Markdown render
+        renderer = CustomRenderer()
+        markdown = mistune.create_markdown(
+            renderer=renderer
+        )
+
+        # resolve markdown
+        markdown(content)
+        
+        # process codeblocks
+        for block in extracted_blocks:
+            artifact_id = str(uuid.uuid4())
+            language = block['language']
+            file_suffix = CODE_FILE_EXTENSION_MAP.get(language, "txt")
+            
+            code_blocks.append({
+                "artifact_id": artifact_id,
+                "content": block['content'],
+                "language": language,
+                "file_suffix": file_suffix
+            })
+        
+        return code_blocks
+
+
+class ShellArtifact(CodeArtifact):
+    shell_result: str = Field(default="", description="shell execution result")
+
+    def __init__(self, artifact_type: ArtifactType, content: Any, code_version: str,
+                 code_interceptor_provider: Optional[str] = None,
+                 artifact_id: Optional[str] = None, render_type: Optional[str] = None,
+                 shell_result: str = "", **kwargs):
+
+        code_type = "shell"
+
+        # extract filename
+        filename = self.extract_filename(content)
+        
+        # default set terminal.txt
+        if not filename:
+            filename = "terminal.txt"
+
+        # update metadata
+        metadata = kwargs.get('metadata', {})
+        metadata['filename'] = filename
+
+        # setting code_interceptor_provider
+        if code_interceptor_provider is None:
+            code_interceptor_provider = "shell_interceptor"
+
+        super().__init__(artifact_type, content, code_type, code_version,
+                         code_interceptor_provider, artifact_id, render_type, metadata=metadata, **kwargs)
+        self.shell_result = shell_result
+
+    def execute(self):
+        # todo add
+        pass
+
+class HtmlArtifact(CodeArtifact):
+
+    def __init__(self, content: Any, artifact_id: Optional[str] = None, **kwargs):
+        # Remove artifact_type from kwargs if it exists to avoid conflicts
+        kwargs.pop('artifact_type', None)
+
+        super().__init__(
+            artifact_type=ArtifactType.HTML,
+            content=content,
+            code_type='html',
+            code_version='1.0',
+            artifact_id=artifact_id,
+            **kwargs
+        )
+        content = content.replace("```html", "").replace("```", "")
+        self.content = None
+        self.attachments.append(
+            ArtifactAttachment(filename=f"{artifact_id}.html", content=content, mime_type="text/html")
+        )

aworld/output/observer.py

@@ -0,0 +1,189 @@
+import logging
+import traceback
+from typing import Callable, List, Dict, Any, Optional, Union
+import inspect
+
+from aworld.output.artifact import Artifact
+
+
+class WorkspaceObserver:
+    """Base class for workspace observers"""
+
+    async def on_create(self, workspace_id: str, artifact: Artifact) -> None:
+        pass
+
+    async def on_update(self, workspace_id: str, artifact: Artifact) -> None:
+        pass
+
+    async def on_delete(self, workspace_id: str, artifact: Artifact) -> None:
+        pass
+
+class Handler:
+    """Handler wrapper to support both functions and class methods"""
+    def __init__(self, func: Callable, instance=None, workspace_id: Optional[str] = None, filters: Optional[Dict[str, Any]] = None):
+        self.func = func
+        self.instance = instance  # Class instance if method
+        self.workspace_id = workspace_id  # Specific workspace to monitor
+        self.filters = filters or {}  # Additional filters (e.g., artifact_type)
+    
+    async def __call__(self, artifact: Artifact, **kwargs) -> Any:
+        """Call the handler with appropriate arguments"""
+        # Check if this handler should process the artifact
+        if self.workspace_id and kwargs.get('workspace_id') != self.workspace_id:
+            return None
+            
+        # Check additional filters
+        for key, value in self.filters.items():
+            if key == 'artifact_type':
+                if artifact.artifact_type != value and artifact.artifact_type.value != value:
+                    return None
+            elif key in artifact.metadata and artifact.metadata[key] != value:
+                return None
+
+        # Get function signature to determine what arguments it expects
+        sig = inspect.signature(self.func)
+        param_count = len(sig.parameters)
+        
+        # Call based on whether it's a method or function, and parameter count
+        if self.instance:
+            if param_count == 0:  # Just self
+                return await self.func() if inspect.iscoroutinefunction(self.func) else self.func()
+            elif param_count == 1:  # Self + artifact
+                return await self.func(artifact) if inspect.iscoroutinefunction(self.func) else self.func(artifact)
+            else:  # Self + artifact + kwargs
+                return await self.func(artifact, **kwargs) if inspect.iscoroutinefunction(self.func) else self.func(artifact, **kwargs)
+        else:
+            if param_count == 0:  # No parameters
+                return await self.func() if inspect.iscoroutinefunction(self.func) else self.func()
+            elif param_count == 1:  # Just artifact
+                return await self.func(artifact) if inspect.iscoroutinefunction(self.func) else self.func(artifact)
+            else:  # Artifact + kwargs
+                return await self.func(artifact, **kwargs) if inspect.iscoroutinefunction(self.func) else self.func(artifact, **kwargs)
+
+
+class DecoratorBasedObserver(WorkspaceObserver):
+    """Enhanced decorator-based observer implementation"""
+    def __init__(self):
+        self.create_handlers: List[Handler] = []
+        self.update_handlers: List[Handler] = []
+        self.delete_handlers: List[Handler] = []
+    
+    async def on_create(self, workspace_id: str, artifact: Artifact, **kwargs) -> List[Any]:
+        """Process artifact creation with all handlers"""
+        results = []
+        for handler in self.create_handlers:
+            try:
+                result = await handler(workspace_id=workspace_id, artifact=artifact, **kwargs)
+                if result is not None:
+                    results.append(result)
+            except Exception as e:
+                print(f"Create handler failed:  error is {e}: {traceback.format_exc()}")
+        return results
+    
+    async def on_update(self, artifact: Artifact, **kwargs) -> List[Any]:
+        """Process artifact update with all handlers"""
+        results = []
+        for handler in self.update_handlers:
+            try:
+                result = await handler(artifact, **kwargs)
+                if result is not None:
+                    results.append(result)
+            except Exception as e:
+                print(f"Update handler failed: {e}")
+        return results
+    
+    async def on_delete(self, artifact: Artifact, **kwargs) -> List[Any]:
+        """Process artifact deletion with all handlers"""
+        results = []
+        for handler in self.delete_handlers:
+            try:
+                result = await handler(artifact, **kwargs)
+                if result is not None:
+                    results.append(result)
+            except Exception as e:
+                print(f"Delete handler failed: {e}")
+        return results
+    
+    def register_create_handler(self, func, instance=None, workspace_id=None, filters=None):
+        """Register a handler for artifact creation"""
+        logging.info(f"[📂WORKSPACE]✨ Registering create handler for {func}")
+        self.create_handlers.append(Handler(func, instance, workspace_id, filters))
+        return func
+
+    def un_register_create_handler(self, func, instance=None, workspace_id=None):
+        """Register a handler for artifact creation"""
+        logging.info(f"[📂WORKSPACE] UnRegister create handler for {func}")
+        for handler in self.create_handlers:
+            if handler.func == func:
+                self.create_handlers.remove(handler)
+                logging.info(f"[📂WORKSPACE] UnRegister create handler for {func} success")
+
+    def register_update_handler(self, func, instance=None, workspace_id=None, filters=None):
+        """Register a handler for artifact update"""
+        logging.info(f"[📂WORKSPACE]✨ Registering update handler for {func}")
+        self.update_handlers.append(Handler(func, instance, workspace_id, filters))
+        return func
+        
+    def register_delete_handler(self, func, instance=None, workspace_id=None, filters=None):
+        """Register a handler for artifact deletion"""
+        logging.info(f"[📂WORKSPACE]✨ Registering delete handler for {func}")
+        self.delete_handlers.append(Handler(func, instance, workspace_id, filters))
+        return func
+
+# Global observer instance
+_observer = DecoratorBasedObserver()
+
+def get_observer() -> DecoratorBasedObserver:
+    """Get the global observer instance"""
+    return _observer
+
+def on_artifact_create(func=None, workspace_id=None, filters=None):
+    """
+    Decorator for artifact creation handlers
+    
+    Can be used as a simple decorator (@on_artifact_create) or with parameters:
+    @on_artifact_create(workspace_id='abc', filters={'artifact_type': 'WEB_PAGES'})
+    """
+    if func is None:
+        # Called with parameters
+        def decorator(f):
+            return _observer.register_create_handler(f, None, workspace_id, filters)
+        return decorator
+    
+    # Called as simple decorator
+    return _observer.register_create_handler(func)
+
+def on_artifact_update(func=None, workspace_id=None, filters=None):
+    """
+    Decorator for artifact update handlers
+    
+    Can be used as a simple decorator (@on_artifact_update) or with parameters:
+    @on_artifact_update(workspace_id='abc', filters={'artifact_type': 'WEB_PAGES'})
+    """
+    if func is None:
+        # Called with parameters
+        def decorator(f):
+            return _observer.register_update_handler(f, None, workspace_id, filters)
+        return decorator
+    
+    # Called as simple decorator
+    return _observer.register_update_handler(func)
+
+def on_artifact_delete(func=None, workspace_id=None, filters=None):
+    """
+    Decorator for artifact deletion handlers
+    
+    Can be used as a simple decorator (@on_artifact_delete) or with parameters:
+    @on_artifact_delete(workspace_id='abc', filters={'artifact_type': 'WEB_PAGES'})
+    """
+    if func is None:
+        # Called with parameters
+        def decorator(f):
+            return _observer.register_delete_handler(f, None, workspace_id, filters)
+        return decorator
+    
+    # Called as simple decorator
+    return _observer.register_delete_handler(func)
+
+
+

aworld/output/outputs.py

@@ -0,0 +1,202 @@
+import abc
+import asyncio
+from abc import abstractmethod
+from dataclasses import field, dataclass
+from typing import AsyncIterator, Any, Union, Iterator
+
+from aworld.logs.util import logger
+from aworld.output import Output
+from aworld.output.base import RUN_FINISHED_SIGNAL
+
+
+@dataclass
+class Outputs(abc.ABC):
+    """Base class for managing output streams in the AWorld framework.
+    Provides abstract methods for adding and streaming outputs both synchronously and asynchronously.
+    reference: https://github.com/openai/openai-agents-python/blob/main/src/agents/result.py
+    """
+    _metadata: dict = field(default_factory=dict)
+
+    @abstractmethod
+    async def add_output(self, output: Output):
+        """Add an output asynchronously to the output stream.
+        
+        Args:
+            output (Output): The output to be added
+        """
+        pass
+
+    @abstractmethod
+    def sync_add_output(self, output: Output):
+        """Add an output synchronously to the output stream.
+        
+        Args:
+            output (Output): The output to be added
+        """
+        pass
+
+    @abstractmethod
+    async def stream_events(self) -> Union[AsyncIterator[Output], list]:
+        """Stream outputs asynchronously.
+        
+        Returns:
+            AsyncIterator[Output]: An async iterator of outputs
+        """
+        pass
+
+    @abstractmethod
+    def sync_stream_events(self) -> Union[Iterator[Output], list]:
+        """Stream outputs synchronously.
+        
+        Returns:
+            Iterator[Output]: An iterator of outputs
+        """
+        pass
+
+    @abstractmethod
+    async def mark_completed(self):
+        pass
+
+    async def get_metadata(self) -> dict:
+        return self._metadata
+
+    async def set_metadata(self, metadata: dict):
+        self._metadata = metadata
+
+@dataclass
+class AsyncOutputs(Outputs):
+    """Intermediate class that implements the Outputs interface with async support.
+    This class serves as a base for more specific async output implementations."""
+
+    async def add_output(self, output: Output):
+        pass
+
+    def sync_add_output(self, output: Output):
+        pass
+
+    async def stream_events(self) -> Union[AsyncIterator[Output], list]:
+        pass
+
+    def sync_stream_events(self) -> Union[Iterator[Output]]:
+        pass
+
+
+@dataclass
+class DefaultOutputs(Outputs):
+    """DefaultAsyncOutputs """
+
+    _outputs: list = field(default_factory=list)
+
+    async def add_output(self, output: Output):
+        self._outputs.append(output)
+
+    def sync_add_output(self, output: Output):
+        self._outputs.append(output)
+
+    async def stream_events(self) -> Union[AsyncIterator[Output], list]:
+        return self._outputs
+
+    def sync_stream_events(self) -> Union[Iterator[Output], list]:
+        return self._outputs
+
+    async def mark_completed(self):
+        pass
+
+
+@dataclass
+class StreamingOutputs(AsyncOutputs):
+    """Concrete implementation of AsyncOutputs that provides streaming functionality.
+    Manages a queue of outputs and handles streaming with error checking and task management."""
+
+    # Task and input related fields
+    # task: Task = Field(default=None)  # The task associated with these outputs
+    input: Any = field(default=None)  # Input data for the task
+    usage: dict = field(default=None)  # Usage statistics
+
+    # State tracking
+    is_complete: bool = field(default=False)  # Flag indicating if streaming is complete
+
+    # Queue for managing outputs
+    _output_queue: asyncio.Queue[Output] = field(
+        default_factory=asyncio.Queue, repr=False
+    )
+
+    # Internal state management
+    _visited_outputs: list[Output] = field(default_factory=list)
+    _stored_exception: Exception | None = field(default=None, repr=False)  # Stores any exceptions that occur
+    _run_impl_task: asyncio.Task[Any] | None = field(default=None, repr=False)  # The running task
+
+    async def add_output(self, output: Output):
+        """Add an output to the queue asynchronously.
+        
+        Args:
+            output (Output): The output to be added to the queue
+        """
+        self._output_queue.put_nowait(output)
+
+    async def stream_events(self) -> AsyncIterator[Output]:
+        """Stream outputs asynchronously, handling cached outputs and new outputs from the queue.
+        Includes error checking and task cleanup.
+        
+        Yields:
+            Output: The next output in the stream
+            
+        Raises:
+            Exception: Any stored exception that occurred during streaming
+        """
+        # First yield any cached outputs
+        for output in self._visited_outputs:
+            if output == RUN_FINISHED_SIGNAL:
+                self._output_queue.task_done()
+                return
+            yield output
+
+        # Main streaming loop
+        while True:
+            self._check_errors()
+            if self._stored_exception:
+                logger.debug("Breaking due to stored exception")
+                self.is_complete = True
+                break
+
+            if self.is_complete and self._output_queue.empty():
+                break
+
+            try:
+                output = await self._output_queue.get()
+                self._visited_outputs.append(output)
+
+            except asyncio.CancelledError:
+                break
+
+            if output == RUN_FINISHED_SIGNAL:
+                self._output_queue.task_done()
+                self._check_errors()
+                break
+
+            yield output
+            self._output_queue.task_done()
+
+        self._cleanup_tasks()
+
+        if self._stored_exception:
+            raise self._stored_exception
+
+    def _check_errors(self):
+        """Check for errors in the streaming process.
+        Verifies step count and checks for exceptions in the running task.
+        """
+        # Check the task for any exceptions
+        if self._run_impl_task and self._run_impl_task.done():
+            exc = self._run_impl_task.exception()
+            if exc and isinstance(exc, Exception):
+                self._stored_exception = exc
+
+    def _cleanup_tasks(self):
+        """Clean up any running tasks by cancelling them if they're not done."""
+        if self._run_impl_task and not self._run_impl_task.done():
+            self._run_impl_task.cancel()
+
+    async def mark_completed(self) -> None:
+        """Mark the streaming process as completed by adding a RUN_FINISHED_SIGNAL to the queue."""
+        await self._output_queue.put(RUN_FINISHED_SIGNAL)

aworld/output/utils.py

@@ -0,0 +1,80 @@
+import os
+from typing import AsyncGenerator, Generator, Callable, Any
+
+from aworld.output.workspace import WorkSpace
+from aworld.output.base import OutputPart, MessageOutput, Output
+
+
+async def consume_output(__output__, callback):
+    if isinstance(__output__, Output):
+        ## parts
+        if __output__.parts:
+            for part in __output__.parts:
+                await consume_part(part, callback)
+        ## MessageOutput
+        if isinstance(__output__, MessageOutput):
+            if __output__.reason_generator or __output__.response_generator:
+                if __output__.reason_generator:
+                    await consume_content(__output__.reason_generator, callback)
+                if __output__.reason_generator:
+                    await consume_content(__output__.response_generator, callback)
+            else:
+                await consume_content(__output__.reasoning, callback)
+                await consume_content(__output__.response, callback)
+            if __output__.tool_calls:
+                await consume_content(__output__.tool_calls, callback)
+        else:
+            await consume_content(__output__.data, callback)
+
+
+
+
+async def consume_part(part, callback):
+    if isinstance(part.content, Output):
+        await consume_output(__output__=part.content, callback=callback)
+    else:
+        await consume_content(__content__=part.content, callback=callback)
+
+
+
+async def consume_content(__content__, callback: Callable[..., Any]):
+    if not __content__:
+        return
+    if isinstance(__content__, AsyncGenerator):
+        async for sub_content in __content__:
+            if isinstance(sub_content, OutputPart):
+                await consume_part(sub_content, callback)
+            elif isinstance(sub_content, Output):
+                await consume_output(sub_content, callback)
+            else:
+                await callback(sub_content)
+    elif isinstance(__content__, Generator) or isinstance(__content__, list):
+        for sub_content in __content__:
+            if isinstance(sub_content, OutputPart):
+                await consume_part(sub_content, callback)
+            elif isinstance(sub_content, Output):
+                await consume_output(sub_content, callback)
+            else:
+                await callback(sub_content)
+    elif isinstance(__content__, str):
+        await callback(__content__)
+    else:
+        await callback(__content__)
+
+
+async def load_workspace(workspace_id: str, workspace_type: str, workspace_parent_path: str):
+    """
+    This function is used to get the workspace by its id.
+    It first checks the workspace type and then creates the workspace accordingly.
+    If the workspace type is not valid, it raises an HTTPException.
+    """
+    if workspace_id is None:
+        raise RuntimeError("workspace_id is None")
+
+    if workspace_type == "local":
+        workspace = WorkSpace.from_local_storages(workspace_id, storage_path=os.path.join(workspace_parent_path, workspace_id))
+    elif workspace_type == "oss":
+        workspace = WorkSpace.from_oss_storages(workspace_id, storage_path=os.path.join(workspace_parent_path, workspace_id))
+    else:
+        raise RuntimeError("Invalid workspace type")
+    return workspace

aworld/output/workspace.py

@@ -0,0 +1,522 @@
+import logging
+import os
+import traceback
+import uuid
+from datetime import datetime
+from typing import Dict, Any, Optional, List, Union
+
+from pydantic import BaseModel, Field, ConfigDict
+
+from aworld.output.artifact import ArtifactType, Artifact
+from aworld.output.code_artifact import CodeArtifact
+from aworld.output.storage.artifact_repository import ArtifactRepository, LocalArtifactRepository
+from aworld.output.observer import WorkspaceObserver, get_observer
+from aworld.output.storage.oss_artifact_repository import OSSArtifactRepository
+
+
+class WorkSpace(BaseModel):
+    """
+    Artifact workspace, managing a group of related artifacts
+    
+    Provides collaborative editing features, supporting version management, update notifications, etc. for multiple Artifacts
+    """
+
+    workspace_id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="unique identifier for the workspace")
+    name: str = Field(default="", description="name of the workspace")
+    created_at: str = Field(default_factory=lambda: datetime.now().isoformat())
+    updated_at: str = Field(default_factory=lambda: datetime.now().isoformat())
+    metadata: Dict[str, Any] = Field(default={}, description="metadata")
+    artifacts: List[Artifact] = Field(default=[], description="list of artifacts")
+
+    artifact_id_index: Dict[str, int] = Field(default={}, description="artifact id index", exclude=True)
+    observers: Optional[List[WorkspaceObserver]] = Field(default=[], description="list of observers", exclude=True)
+    repository: Optional[ArtifactRepository] = Field(default=None, description="local artifact repository", exclude=True)
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+    
+    def __init__(
+            self,
+            workspace_id: Optional[str] = None,
+            name: Optional[str] = None,
+            storage_path: Optional[str] = None,
+            observers: Optional[List[WorkspaceObserver]] = None,
+            use_default_observer: bool = True,
+            clear_existing: bool = False,
+            repository: Optional[ArtifactRepository] = None
+    ):
+        super().__init__()
+        self.workspace_id = workspace_id or str(uuid.uuid4())
+        self.name = name or f"Workspace-{self.workspace_id[:8]}"
+        self.created_at = datetime.now().isoformat()
+        self.updated_at = self.created_at
+
+        # Initialize repository first
+        storage_dir = storage_path or os.path.join("data", "workspaces", self.workspace_id)
+        if repository is None:
+            self.repository = LocalArtifactRepository(storage_dir)
+        else:
+            self.repository = repository
+
+        # Initialize artifacts and metadata
+        if clear_existing:
+            self.artifacts = []
+            self.metadata = {}
+        else:
+            # Try to load existing workspace data
+            workspace_data = self._load_workspace_data()
+            if workspace_data:
+                self.artifacts = workspace_data.get('artifacts', [])
+                self.metadata = workspace_data.get('metadata', {})
+                self.created_at = workspace_data.get('created_at', self.created_at)
+                self.updated_at = workspace_data.get('updated_at', self.updated_at)
+            else:
+                self.artifacts = []
+                self.metadata = {}
+
+        # Build artifact_id_index after loading artifacts
+        self._rebuild_artifact_id_index()
+
+        # Initialize observers
+        self.observers: List[WorkspaceObserver] = []
+        if use_default_observer:
+            self.observers.append(get_observer())
+
+        if observers:
+            for observer in observers:
+                if observer not in self.observers:  # Avoid duplicates
+                    self.add_observer(observer)
+
+    def _load_workspace_data(self) -> Optional[Dict[str, Any]]:
+        """
+        Load workspace data from repository
+        
+        Returns:
+            Dictionary containing workspace data if exists, None otherwise
+        """
+        try:
+            # Get workspace versions
+
+            workspace_data = self.repository.load_index()
+
+            if not workspace_data:
+                return None
+
+            # Load artifacts
+            artifacts = []
+            # First load the artifacts list from workspace data
+            workspace_artifacts = workspace_data.get("artifacts", [])
+            for artifact_data in workspace_artifacts:
+                artifact_id = artifact_data.get("artifact_id")
+                if artifact_id:
+                    artifact_data = self.repository.retrieve_latest_artifact(artifact_id)
+                    if artifact_data:
+                        artifacts.append(Artifact.from_dict(artifact_data))
+
+            return {
+                "artifacts": artifacts,
+                "metadata": workspace_data.get("metadata", {}),
+                "created_at": workspace_data.get("created_at"),
+                "updated_at": workspace_data.get("updated_at")
+            }
+        except Exception as e:
+            traceback.print_exc()
+            print(f"Error loading workspace data: {e}")
+            return None
+
+    @classmethod
+    def from_local_storages(cls, workspace_id: Optional[str] = None,
+                            name: Optional[str] = None,
+                            storage_path: Optional[str] = None,
+                            observers: Optional[List[WorkspaceObserver]] = None,
+                            use_default_observer: bool = True
+                            ) -> "WorkSpace":
+        """
+        Create a workspace instance from local storage
+        
+        Args:
+            workspace_id: Optional workspace ID
+            name: Optional workspace name
+            storage_path: Optional storage path
+            observers: Optional list of observers
+            use_default_observer: Whether to use default observer
+            
+        Returns:
+            WorkSpace instance
+        """
+        if storage_path is None:
+            storage_path = os.path.join("data", "workspaces", workspace_id)
+        workspace = cls(
+            workspace_id=workspace_id,
+            name=name,
+            storage_path=storage_path,
+            observers=observers,
+            use_default_observer=use_default_observer,
+            clear_existing=False  # Always try to load existing data
+        )
+        return workspace
+
+    @classmethod
+    def from_oss_storages(cls,
+                          workspace_id: Optional[str] = None,
+                          name: Optional[str] = None,
+                          storage_path: Optional[str] = "aworld/workspaces/",
+                          observers: Optional[List[WorkspaceObserver]] = None,
+                          use_default_observer: bool = True,
+                          oss_config: Optional[Dict[str, Any]] = None,
+                          ) -> "WorkSpace":
+        if oss_config is None:
+            oss_config = {
+                "access_key_id": os.getenv("OSS_ACCESS_KEY_ID"),
+                "access_key_secret": os.getenv("OSS_ACCESS_KEY_SECRET"),
+                "endpoint": os.getenv("OSS_ENDPOINT"),
+                "bucket_name": os.getenv("OSS_BUCKET_NAME"),
+            }
+        repository = OSSArtifactRepository(
+            access_key_id=oss_config["access_key_id"],
+            access_key_secret=oss_config["access_key_secret"],
+            endpoint=oss_config["endpoint"],
+            bucket_name=oss_config["bucket_name"],
+            storage_path=storage_path
+        )
+        workspace = cls(
+            workspace_id=workspace_id,
+            name=name,
+            storage_path=storage_path,
+            observers=observers,
+            use_default_observer=use_default_observer,
+            repository=repository
+        )
+        return workspace
+
+    async def create_artifact(
+            self,
+            artifact_type: Union[ArtifactType, str],
+            artifact_id: Optional[str] = None,
+            content: Optional[Any] = None,
+            metadata: Optional[Dict[str, Any]] = None
+    ) -> List[Artifact]:
+        """
+        Create a new artifact
+
+        Args:
+            artifact_type: Artifact type (enum or string)
+            artifact_id: Optional artifact ID (will be generated if not provided)
+            content: Artifact content
+            metadata: Metadata dictionary
+
+        Returns:
+            List of created artifact objects
+        """
+        # If a string is passed, convert to enum type
+        if isinstance(artifact_type, str):
+            artifact_type = ArtifactType(artifact_type)
+
+        # Create new artifacts
+        artifacts = []
+
+        # Ensure metadata is a dictionary
+        if metadata is None:
+            metadata = {}
+
+        # Ensure artifact_id is a valid string
+        if artifact_id is None:
+            artifact_id = str(uuid.uuid4())
+
+        if artifact_type == ArtifactType.CODE:
+            artifacts = CodeArtifact.from_code_content(artifact_type, content)
+        else:
+            artifact = Artifact(
+                artifact_id=artifact_id,
+                artifact_type=artifact_type,
+                content=content,
+                metadata=metadata
+            )
+            artifacts.append(artifact)  # Add single artifact to the list
+
+        # Add to workspace
+        for artifact in artifacts:
+            # Store in repository
+            await self._store_artifact(artifact)
+            logging.info(f"[📂WORKSPACE]💾 Storing artifact in repository: {artifact.artifact_id}")
+
+
+        # Update workspace time
+        self.updated_at = datetime.now().isoformat()
+
+        # Save workspace state to create new version
+        self.save()
+
+        return artifacts  # Return the list of created artifacts
+
+    async def add_artifact(
+            self,
+            artifact: Artifact
+    ) -> None:
+        """
+        Create a new artifact
+
+        Args:
+            artifact: Artifact
+
+        Returns:
+            List of created artifact objects
+        """
+
+        # Store in repository
+        await self._store_artifact(artifact)
+
+        # Update workspace time
+        self.updated_at = datetime.now().isoformat()
+
+        # Save workspace state to create new version
+        self.save()
+
+        await self._notify_observers("create", artifact)
+
+    async def mark_as_completed(self, artifact_id: str) -> None:
+        """
+        Mark an artifact as completed
+        """
+        artifact = self.get_artifact(artifact_id)
+        if artifact:
+            artifact.mark_complete()
+            self.repository.store_artifact(artifact)
+            logging.info(f"[📂WORKSPACE]🎉 Marking artifact as completed: {artifact_id}")
+            await self._notify_observers("complete", artifact)
+        self.save()
+
+    def get_artifact(self, artifact_id: str) -> Optional[Artifact]:
+        """Get artifact with the specified ID"""
+        for artifact in self.artifacts:
+            if artifact.artifact_id == artifact_id:
+                return artifact
+        return None
+
+
+    def get_terminal(self) -> str:
+        pass
+
+    def get_webpage_groups(self) -> list[Any] | None:
+        return self.list_artifacts(ArtifactType.WEB_PAGES)
+
+
+    async def update_artifact(
+            self,
+            artifact_id: str,
+            content: Any,
+            description: str = "Content update"
+    ) -> Optional[Artifact]:
+        """
+        Update artifact content
+        
+        Args:
+            artifact_id: Artifact ID
+            content: New content
+            description: Update description
+            
+        Returns:
+            Updated artifact, or None if it doesn't exist
+        """
+        artifact = self.get_artifact(artifact_id)
+        if artifact:
+            artifact.update_content(content, description)
+
+            # Update storage
+            await self._store_artifact(artifact)
+
+            return artifact
+        return None
+
+    async def delete_artifact(self, artifact_id: str) -> bool:
+        """
+        Delete an artifact from the workspace
+        
+        Args:
+            artifact_id: Artifact ID
+            
+        Returns:
+            Whether deletion was successful
+        """
+        existed = self._check_artifact_exists(artifact_id)
+        if not existed:
+            return True
+        for i, artifact in enumerate(self.artifacts):
+            if artifact.artifact_id == artifact_id:
+                # Remove from list
+                self.artifacts.pop(i)
+
+                # Update workspace time
+                self.updated_at = datetime.now().isoformat()
+
+                self.repository.delete_artifact(artifact_id)
+                # Save workspace state to create new version
+                self.save()
+
+                # Notify observers
+                await self._notify_observers("delete", artifact)
+                return True
+        return False
+
+    def list_artifacts(self, filter_type: Optional[ArtifactType] = None) -> List[Artifact]:
+        """
+        List all artifacts in the workspace
+        
+        Args:
+            filter_type: Optional filter type
+            
+        Returns:
+            List of artifacts
+        """
+        if filter_type:
+            return [a for a in self.artifacts if a.artifact_type == filter_type]
+        return self.artifacts
+
+    def add_observer(self, observer: WorkspaceObserver) -> None:
+        """
+        Add a workspace observer
+        
+        Args:
+            observer: Observer object implementing WorkspaceObserver interface
+        """
+        if not isinstance(observer, WorkspaceObserver):
+            raise TypeError("Observer must be an instance of WorkspaceObserver")
+        self.observers.append(observer)
+
+    def remove_observer(self, observer: WorkspaceObserver) -> None:
+        """Remove an observer"""
+        if observer in self.observers:
+            self.observers.remove(observer)
+
+    async def _notify_observers(self, operation: str, artifact: Artifact) -> List[Any]:
+        """
+        Notify all observers of workspace changes
+        
+        Args:
+            operation: Type of operation (create, update, delete)
+            artifact: Affected artifact
+            
+        Returns:
+            List of results from handlers
+        """
+        results = []
+        for observer in self.observers:
+            try:
+                if operation == "create":
+                    result = await observer.on_create(workspace_id=self.workspace_id, artifact=artifact)
+                    if result:
+                        results.append(result)
+                elif operation == "update":
+                    result = await observer.on_update(workspace_id=self.workspace_id, artifact=artifact)
+                    if result:
+                        results.append(result)
+                elif operation == "delete":
+                    result = await observer.on_delete(workspace_id=self.workspace_id, artifact=artifact)
+                    if result:
+                        results.append(result)
+            except Exception as e:
+                print(f"Observer notification failed: {e}")
+        return results
+    
+    def _check_artifact_exists(self, artifact_id: str) -> bool:
+        return self.artifact_id_index.get(artifact_id, -1) >= 0
+
+
+    def _append_artifact(self, artifact: Artifact) -> None:
+        self.artifacts.append(artifact)
+        logging.info(f"[📂WORKSPACE]🆕 Appending artifact in repository: {artifact.artifact_id}")
+
+
+    def _update_artifact(self, artifact: Artifact) -> None:
+        for i, a in enumerate(self.artifacts):
+            if a.artifact_id == artifact.artifact_id:
+                self.artifacts[i] = artifact
+                logging.info(f"[📂WORKSPACE]🔄 Updating artifact in repository: {artifact.artifact_id}")
+                break
+
+    
+    async def _store_artifact(self, artifact: Artifact) -> None:
+        if self._check_artifact_exists(artifact.artifact_id):
+            self._update_artifact(artifact)
+            await self._notify_observers("update", artifact)
+        else:
+            self._append_artifact(artifact)
+            await self._notify_observers("create", artifact)
+
+        """Store artifact in repository"""
+        artifact_data = artifact.to_dict()
+
+        # Include complete version history
+        artifact_data["version_history"] = artifact.version_history
+
+        version_id = self.repository.store_artifact(artifact)
+
+        # Store in repository
+        artifact.current_version = version_id
+
+    def save(self) -> None:
+        """
+        Save workspace state
+
+        Returns:
+            Workspace storage ID
+        """
+        workspace_data = {
+            "workspace_id": self.workspace_id,
+            "name": self.name,
+            "created_at": self.created_at,
+            "updated_at": self.updated_at,
+            "metadata": self.metadata,
+            "artifact_ids": [a.artifact_id for a in self.artifacts],
+            "artifacts": [
+                {
+                    "artifact_id": a.artifact_id,
+                    "type": str(a.artifact_type),
+                    "metadata": a.metadata,
+                    # "version": a.current_version
+                } for a in self.artifacts
+            ]
+        }
+
+        # Store workspace information with workspace_id in metadata
+        self.repository.save_index(workspace_data)
+        self._rebuild_artifact_id_index()
+
+    def get_file_content_by_artifact_id(self, artifact_id: str) -> str:
+        """
+        Get concatenated content of all artifacts with the same filename.
+        
+        Args:
+            artifact_id: artifact_id
+            
+        Returns:
+            Raw unescaped concatenated content of all matching artifacts
+        """
+        filename = artifact_id
+        for artifact in self.artifacts:
+            if artifact.artifact_id == artifact_id:
+                filename = artifact.metadata.get('filename')
+                break
+
+        result = ""
+        for artifact in self.artifacts:
+            if artifact.metadata.get('filename') == filename:
+                if artifact.content:
+                    result = result + artifact.content
+        decoded_string = result.encode('utf-8').decode('unicode_escape')
+        print(result)
+
+        return decoded_string
+
+    def generate_tree_data(self) -> Dict[str, Any]:
+        """
+        Generate a directory tree structure using the repository's implementation.
+        Returns:
+            A dictionary representing the directory tree.
+        """
+        return self.repository.generate_tree_data(self.name)
+
+    def _rebuild_artifact_id_index(self) -> None:
+        """
+        Rebuild the artifact_id_index mapping artifact_id to its index in self.artifacts.
+        """
+        self.artifact_id_index = {artifact.artifact_id: idx for idx, artifact in enumerate(self.artifacts)}