Spaces:
Sleeping
Sleeping
File size: 3,308 Bytes
47aff61 |
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 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 |
# coding: utf-8
# Copyright (c) 2025 inclusionAI.
import asyncio
from typing import List, Dict, Union
from aworld.config import RunConfig
from aworld.config.conf import TaskConfig
from aworld.agents.llm_agent import Agent
from aworld.core.agent.swarm import Swarm
from aworld.core.common import Config
from aworld.core.task import Task, TaskResponse, Runner
from aworld.output import StreamingOutputs
from aworld.runners.utils import choose_runners, execute_runner
from aworld.utils.common import sync_exec
class Runners:
"""Unified entrance to the utility class of the runnable task of execution."""
@staticmethod
def streamed_run_task(task: Task) -> StreamingOutputs:
"""Run the task in stream output."""
if not task.conf:
task.conf = TaskConfig()
streamed_result = StreamingOutputs(
input=task.input,
usage={},
is_complete=False
)
task.outputs = streamed_result
streamed_result._run_impl_task = asyncio.create_task(
Runners.run_task(task)
)
return streamed_result
@staticmethod
async def run_task(task: Union[Task, List[Task]], run_conf: RunConfig = None) -> Dict[str, TaskResponse]:
"""Run tasks for some complex scenarios where agents cannot be directly used.
Args:
task: User task define.
run_conf:
"""
if isinstance(task, Task):
task = [task]
runners: List[Runner] = await choose_runners(task)
return await execute_runner(runners, run_conf)
@staticmethod
def sync_run_task(task: Union[Task, List[Task]], run_conf: Config = None) -> Dict[str, TaskResponse]:
return sync_exec(Runners.run_task, task=task, run_conf=run_conf)
@staticmethod
def sync_run(
input: str,
agent: Agent = None,
swarm: Swarm = None,
tool_names: List[str] = [],
session_id: str = None
) -> TaskResponse:
return sync_exec(
Runners.run,
input=input,
agent=agent,
swarm=swarm,
tool_names=tool_names,
session_id=session_id
)
@staticmethod
async def run(
input: str,
agent: Agent = None,
swarm: Swarm = None,
tool_names: List[str] = [],
session_id: str = None
) -> TaskResponse:
"""Run agent directly with input and tool names.
Args:
input: User query.
agent: An agent with AI model configured, prompts, tools, mcp servers and other agents.
swarm: Multi-agent topo.
tool_names: Tool name list.
session_id: Session id.
Returns:
TaskResponse: Task response.
"""
if agent and swarm:
raise ValueError("`agent` and `swarm` only choose one.")
if not input:
raise ValueError('`input` is empty.')
if agent:
agent.task = input
swarm = Swarm(agent)
task = Task(input=input, swarm=swarm, tool_names=tool_names,
event_driven=swarm.event_driven, session_id=session_id)
res = await Runners.run_task(task)
return res.get(task.id)
|