Elkar AI 任务管理开源项目教程

1. 项目介绍

Elkar 是一个基于 Google 的 Agent2Agent 协议（A2A）的开源任务管理工具。它旨在帮助开发者构建协作、自主的多代理系统，而无需担心管理基础设施的复杂性。Elkar 提供了一个 Python SDK（其他语言的支持即将到来），用于构建和使用 A2A 协议的 AI 代理，并具有存储和队列管理功能。此外，它还提供了一个应用程序和 API，用于管理、监控和调试代理任务。

2. 项目快速启动

以下是快速启动 Elkar 项目的步骤：

步骤 1: 注册并获取 API Key

1. 在 Elkar 平台上创建一个账户。
2. 创建一个新的代理：导航到“代理”>“添加新代理”。
3. 生成 API Key。注意：现在复制 API key —— 它不会再次显示。

步骤 2: 安装 Elkar SDK

运行以下命令安装 Elkar SDK：

pip install elkar

步骤 3: 创建任务处理器

任务处理器管理任务的状态和工件，抽象化与任务存储的交互复杂性。它支持三种操作：设置状态（set_status）、将消息添加到历史记录（add_messages_to_history）和插入或更新工件（upsert_artifact）。

下面是一个任务处理器的示例：

from elkar.a2a_types import *
from elkar.server.server import A2AServer
from elkar.task_manager.task_manager_base import RequestContext
from elkar.task_manager.task_manager_with_task_modifier import TaskManagerWithModifier
from elkar.task_modifier.base import TaskModifierBase

async def task_handler(task: TaskModifierBase, request_context: RequestContext | None) -> None:
    await task.set_status(
        TaskStatus(
            state=TaskState.WORKING,
            message=Message(
                role="agent",
                parts=[
                    TextPart(text="我理解了任务，我正在处理...")
                ]
            )
        )
    )
    await task.upsert_artifacts(
        [
            Artifact(
                parts=[
                    TextPart(text="我已经完成了任务，这是结果...")
                ],
                index=0
            )
        ]
    )
    await task.set_status(
        TaskStatus(
            state=TaskState.COMPLETED,
            message=Message(
                role="agent",
                parts=[
                    TextPart(text="我已经完成了任务！")
                ]
            ),
            is_final=True
        )
    )

步骤 4: 创建代理卡片

代理卡片是根据 A2A 协议的要求定义的。

agent_card = AgentCard(
    name="测试代理",
    description="测试代理描述",
    url="https://example.com",
    provider=AgentProvider(
        organization="Elkar",
        url="https://www.elkar.co"
    ),
    documentationUrl="https://example.com/documentation",
    version="1.0.0",
    skills=[
        AgentSkill(
            id="1",
            name="生成图像",
            description="根据提示描述生成图像",
            inputModes=["text"],
            outputModes=["image"]
        )
    ],
    capabilities=AgentCapabilities(
        streaming=True,
        pushNotifications=True,
        stateTransitionHistory=True
    )
)

步骤 5: 创建 A2A 服务器

使用托管服务作为任务存储实例化您的 A2A 服务器。

api_key = "YOUR_ELKAR_API_KEY"  # 替换为您的实际 Elkar API key
store = ElkarClientStore(
    base_url="https://api.elkar.co/api",
    api_key=api_key
)
task_manager = TaskManagerWithModifier(
    agent_card,
    send_task_handler=task_handler,
    store=store
)
# 创建服务器实例
server = A2AServer(
    task_manager,
    host="0.0.0.0",
    port=5001,
    endpoint="/"
)
server.start()  # 这是阻塞的。在生产环境中，使用像 Uvicorn 这样的 ASGI 服务器。

3. 应用案例和最佳实践

• 任务跟踪与管理：Elkar 可以轻松跟踪和管理长时间运行的任务，并提供异步操作的支持。
• 任务历史浏览：使用 Elkar，您可以浏览和管理任务历史，以便观察和调试。
• 实时任务流：Elkar 允许通过专用的 SDK 实时在代理之间流动任务。
• 深度调试：利用 Elkar，您可以深度调试代理任务和 A2A 服务器交互，并获得任务历史、工件和服务器通信的完全可见性。

4. 典型生态项目

Elkar 作为任务管理工具，可以集成到各种生态项目中，例如：

• 自动化工作流：集成到 CI/CD 流程中，自动化任务分配和结果收集。
• 多代理协作：构建多代理系统，实现复杂的任务分解和协作。
• 智能助手集成：集成到智能助手中，提供任务管理和调度功能。

以上就是 Elkar AI 任务管理开源项目的教程，希望对您的开发工作有所帮助。