97 lines
3.0 KiB
Python
97 lines
3.0 KiB
Python
|
import logging
|
|||
|
|
|
|||
|
|
from a2a.server.agent_execution import AgentExecutor, RequestContext
|
|||
|
|
from a2a.server.events import EventQueue
|
|||
|
|
from a2a.server.tasks import TaskUpdater
|
|||
|
|
from a2a.types import (
|
|||
|
|
InternalError,
|
|||
|
|
InvalidParamsError,
|
|||
|
|
Part,
|
|||
|
|
TaskState,
|
|||
|
|
TextPart,
|
|||
|
|
UnsupportedOperationError,
|
|||
|
|
)
|
|||
|
|
from a2a.utils import (
|
|||
|
|
new_agent_text_message,
|
|||
|
|
new_task,
|
|||
|
|
)
|
|||
|
|
from a2a.utils.errors import ServerError
|
|||
|
|
|
|||
|
|
from agent import ConductorAgent
|
|||
|
|
|
|||
|
|
|
|||
|
|
logging.basicConfig(level=logging.INFO)
|
|||
|
|
logger = logging.getLogger(__name__)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class ConductorAgentExecutor(AgentExecutor):
|
|||
|
|
"""Conductor AgentExecutor - 总管理代理执行器."""
|
|||
|
|
|
|||
|
|
def __init__(self):
|
|||
|
|
self.agent = ConductorAgent()
|
|||
|
|
|
|||
|
|
async def execute(
|
|||
|
|
self,
|
|||
|
|
context: RequestContext,
|
|||
|
|
event_queue: EventQueue,
|
|||
|
|
) -> None:
|
|||
|
|
error = self._validate_request(context)
|
|||
|
|
if error:
|
|||
|
|
raise ServerError(error=InvalidParamsError())
|
|||
|
|
|
|||
|
|
query = context.get_user_input()
|
|||
|
|
task = context.current_task
|
|||
|
|
if not task:
|
|||
|
|
task = new_task(context.message) # type: ignore
|
|||
|
|
await event_queue.enqueue_event(task)
|
|||
|
|
updater = TaskUpdater(event_queue, task.id, task.context_id)
|
|||
|
|
try:
|
|||
|
|
# 使用非流式invoke方法
|
|||
|
|
result = await self.agent.invoke(query, task.context_id)
|
|||
|
|
|
|||
|
|
is_task_complete = result.get('is_task_complete', True)
|
|||
|
|
require_user_input = result.get('require_user_input', False)
|
|||
|
|
content = result.get('content', '处理完成')
|
|||
|
|
|
|||
|
|
if require_user_input:
|
|||
|
|
await updater.update_status(
|
|||
|
|
TaskState.input_required,
|
|||
|
|
new_agent_text_message(
|
|||
|
|
content,
|
|||
|
|
task.context_id,
|
|||
|
|
task.id,
|
|||
|
|
),
|
|||
|
|
final=True,
|
|||
|
|
)
|
|||
|
|
elif is_task_complete:
|
|||
|
|
await updater.add_artifact(
|
|||
|
|
[Part(root=TextPart(text=content))],
|
|||
|
|
name='conductor_result',
|
|||
|
|
)
|
|||
|
|
await updater.complete()
|
|||
|
|
else:
|
|||
|
|
# 如果既不需要输入也未完成，设置为working状态
|
|||
|
|
await updater.update_status(
|
|||
|
|
TaskState.working,
|
|||
|
|
new_agent_text_message(
|
|||
|
|
content,
|
|||
|
|
task.context_id,
|
|||
|
|
task.id,
|
|||
|
|
),
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f'An error occurred while processing the request: {e}')
|
|||
|
|
raise ServerError(error=InternalError()) from e
|
|||
|
|
|
|||
|
|
def _validate_request(self, context: RequestContext) -> bool:
|
|||
|
|
# 这里可以添加请求验证逻辑
|
|||
|
|
# 返回 True 表示有错误，False 表示验证通过
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
async def cancel(
|
|||
|
|
self, context: RequestContext, event_queue: EventQueue
|
|||
|
|
) -> None:
|
|||
|
|
raise ServerError(error=UnsupportedOperationError())
|
|||
|
|
|