workflows 库内置一个 WorkflowServer 类，让你轻松把工作流通过 HTTP API 暴露出去。这为从任何支持 HTTP 的客户端运行与管理工作流提供了一种灵活的方式。

此外，WorkflowServer 还会一并部署一个静态调试应用，让你可视化、运行并调试工作流。它自动挂载在运行服务的根路径 / 上。

安装

该工作流服务端是独立于核心 llama-index-workflows 库的包：

终端窗口

pip install llama-agents-server

编程式使用

你可以创建服务端、加入自己的工作流，并以编程方式运行它。当你想把工作流服务端嵌入一个更大的应用时，这很有用。

首先创建一个 Python 文件（例如 my_server.py）：

my_server.py

import asyncio
from workflows import Workflow, step
from workflows.context import Context
from workflows.events import Event, StartEvent, StopEvent
from llama_agents.server import WorkflowServer

class StreamEvent(Event):
    sequence: int

# Define a simple workflow
class GreetingWorkflow(Workflow):
    @step
    async def greet(self, ctx: Context, ev: StartEvent) -> StopEvent:
        for i in range(3):
            ctx.write_event_to_stream(StreamEvent(sequence=i))
            await asyncio.sleep(0.3)

        name = ev.get("name", "World")
        return StopEvent(result=f"Hello, {name}!")

greet_wf = GreetingWorkflow()

# Create a server instance
server = WorkflowServer()

# Add the workflow to the server
server.add_workflow("greet", greet_wf)

# To run the server programmatically (e.g., from your own script)
# import asyncio
#
# async def main():
#     await server.serve(host="0.0.0.0", port=8080)
#
# if __name__ == "__main__":
#     asyncio.run(main())

命令行工具（CLI）

该库还提供了一个便捷的 CLI，可以从包含 WorkflowServer 实例的文件启动服务。

以上例中的 my_server.py 为例，你可以用以下命令启动服务：

终端窗口

python -m llama_agents.server my_server.py

服务默认监听 0.0.0.0:8080。你可以用 WORKFLOWS_PY_SERVER_HOST 与 WORKFLOWS_PY_SERVER_PORT 环境变量来配置主机与端口。

工作流调试器 UI

此外，WorkflowServer 还会一并部署一个静态调试应用，让你可视化、运行并调试工作流。它自动挂载在运行服务的根路径 / 上。

工作流调试 UI 提供几项关键能力：

• 工作流可视化：UI 会给出工作流结构的可视化表示，静态图与运行中的动态图都有。你还可以按需重新排布节点。
• 自动 schema 探测：如果你定制了起始/结束事件或内部事件的 schema，UI 会自动探测并展示与之一致的界面。
• 人在环路：工作流运行期间，你可以向它发送任意事件。对于依赖人工输入才能继续执行的工作流很有用。参见事件日志顶部的 Send Event 按钮。
• 事件日志：所有流式事件都会记录在 UI 中，让你在右侧面板实时查看工作流的执行过程。
• 多次运行：调试并比较多次运行。每运行一次工作流，左侧面板就会记录一次。
• 多个工作流：UI 允许你运行挂载到该 WorkflowServer 上的任意工作流。

处理“隐藏”事件

有时，工作流会发送/接收那些在工作流中以注解声明的事件（比如用 ctx.wait_for_event()）。这种情况下，你仍然可以用 server.add_workflow(..., additional_events=[...]) 把这些事件告知 UI。这样 Send Event 等界面元素就能识别这些事件。

API 端点

WorkflowServer 暴露以下 RESTful 端点：

方法	路径	说明
GET	/health	返回健康检查响应（{"status": "healthy"}）。
GET	/workflows	列出所有已注册工作流的名称。
POST	/workflows/{name}/run	同步运行指定工作流并返回最终结果。
POST	/workflows/{name}/run-nowait	异步启动指定工作流并返回 handler_id。
GET	/handlers/{handler_id}	获取异步运行的工作流结果。仍在运行时返回 202 Accepted，失败返回 500，完成返回 200。
GET	/events/{handler_id}	以按行分隔的 JSON（启用 SSE 时为 text/event-stream）流式输出运行中工作流的全部事件（`application/x-ndjson）。
POST	/events/{handler_id}	在工作流执行期间向其发送一个事件（对人在环路很有用）
GET	/handlers	获取全部工作流 handler（运行中与已完成的）
POST	/handlers/{handler_id}/cancel	停止并取消工作流的执行。

运行工作流（/run）

要运行工作流并等待其完成，向 /workflows/{name}/run 发送 POST 请求。

请求体：

{
  "start_event": {},
  "context": {},
  "handler_id": "",
}

• start_event：StartEvent 或其子类的序列化表示。推荐用它作为工作流的输入。
• context：工作流上下文的序列化表示
• handler_id：工作流 handler 标识，用于从上一次已完成的运行继续。

成功响应（200 OK）：

{
  "result": "The workflow has been successfully run"
}

异步运行工作流（/run-nowait）

要启动工作流而不等待它结束，使用 /run-nowait 端点。

请求体：

{
  "start_event": {},
  "context": {},
  "handler_id": ""
}

请求体的参数与 /run 端点相同。

成功响应（200 OK）：

{
  "handler_id": "someUniqueId123",
  "status": "started"
}

随后你可以用这个 handler_id 查看结果或流式接收事件。

流式事件（GET /events/{handler_id}）

该端点仅在你先前用 /run-nowait 异步启动过工作流时可用

要以 Server-Sent Events（SSE）或多行 JSON 的形式流式接收事件，可以向 /events/{handler_id} 端点发请求，并带上你先前启动的异步工作流运行的 handler ID。

查询参数

• sse（取“true”或“false”，非必填）：为 true 时以 Server Sent Events（text/event-stream）流式输出，否则以多行 JSON 输出（application/x-ndjson）。默认为 true。
• acquire_timeout（可转 float 的字符串，非必填）：获取锁以遍历事件的超时时间
• include_internal（取“true”或“false”，非必填）：为 true 时流式输出工作流内部事件。默认为 false。
• include_qualified_name（取“true”或“false”，非必填）：在响应体中包含事件的限定名。默认为 true。

请求示例

终端窗口

curl http://localhost:80/events/someUniqueId123?sse=false&acquire_timeout=1&include_internal=false&include_qualified_name=true

成功响应（200 OK）

单个事件的载荷：

{
  "value": {"result": 12},
  "qualified_name": "__main__.MathEvent",
  "type": "__main__.MathEvent",
  "types": ["workflows.events.Event", "__main__.MathEvent"],
}

重要注意事项

• 每次工作流运行只允许一个读者流式接收事件
• 事件一旦被流式读出就无法再恢复（除非你在客户端侧实现了某种持久化逻辑）

我们正在改进这两点，因此服务端行为可能还会变化

获取工作流的执行结果（/results/{handler_id}）

该端点仅在你先前用 /run-nowait 异步启动过工作流时可用

要获取先前启动的异步工作流运行的结果，可以调用 /results/{handler_id} 端点并传入该运行的 handler ID。

请求示例

终端窗口

curl http://localhost:80/results/someUniqueId123

成功响应（200 OK）

{
  "handler_id": "someUniqueId123",
  "workflow_name": "math_workflow",
  "run_id": "uniqueRunId456",
  "error": null,
  "result": {
    "sum": 15,
    "subtraction": 9,
    "multiplication": 36,
    "division": 4,
  },
  "status": "completed",
  "started_at": "2024-10-21T14:32:15.123Z",
  "updated_at": "2024-10-21T14:45:30.456Z",
  "completed_at": "2024-10-21T14:45:30.456Z"
}

已接受响应（202 ACCEPTED）

当工作流仍在运行、尚未产生结果时，返回状态码 202。

发送事件（POST /events/{handler_id}）

在工作流需要外部输入才能运行的场合（例如人在环路），你可以向 events/{handler_id} 端点发 POST 请求，带上要发送的事件数据（以及可选的、要送达的那个步骤），以此提供该外部输入。

请求体

{
  "event": {"__is_pydantic": true, "value": {"feedback": "This is great!", "approved": true}, "qualified_name": "__main__.HumanFeedbackEvent"},
  "step": "process_human_feedback"
}

• event：工作流 Event 的序列化表示。
• step（可选）：要发送到的步骤名称。

成功响应（200 OK）

{
  "status": "sent"
}

取消一次工作流运行（/handlers/{handler_id}/cancel）

要停掉一个正在运行的 handler（取消它的任务），并可选地把关联的 handler 从持久化存储中移除，可以使用 /handlers/{handler_id}/cancel。

查询参数

• purge（取“true”或“false”，非必填）：是否把与该工作流关联的 handler 从持久化存储中移除。默认为 false。

请求示例

终端窗口

curl -X POST http://localhost:80/handlers/someUniqueId123/cancel?purge=true

成功响应（200 OK）

{
  "status": "deleted", // or canceled if purge is false
}

持久化

默认情况下，WorkflowServer 使用内存存储（MemoryWorkflowStore），因此进程重启后所有 handler 状态与事件都会丢失。要做持久化，请传入一个由数据库支撑的 workflow_store。

SQLite

单进程部署最简单的选择是 SqliteWorkflowStore，它会把 handler 状态、事件与结果持久化到本地文件：

from llama_agents.server import WorkflowServer, SqliteWorkflowStore

store = SqliteWorkflowStore(db_path="workflows.db")

server = WorkflowServer(workflow_store=store)
server.add_workflow("greet", greet_wf)

DBOS（Postgres）

对于需要 Postgres 持久化、可靠执行以及分布式 worker 的生产部署，请使用 llama-agents-dbos 包中的 DBOSRuntime。它用 DBOS 支撑的实现替换默认运行时，提供跨进程重启的事务性状态管理与恢复：

from dbos import DBOS
from llama_agents.dbos import DBOSRuntime
from llama_agents.server import WorkflowServer

# Configure DBOS. It uses SQLite by default, or set system_database_url for Postgres.
DBOS(config={"name": "my-app", "run_admin_server": False})

runtime = DBOSRuntime()

server = WorkflowServer(
    workflow_store=runtime.create_workflow_store(),
    runtime=runtime.build_server_runtime(),
)
server.add_workflow("greet", GreetingWorkflow())

默认情况下 DBOS 使用 SQLite（零配置）。要使用 Postgres，请在 DBOS 配置中传入 system_database_url。多副本部署时，每个副本必须有唯一的 executor_id：

DBOS(config={
    "name": "my-app",
    "system_database_url": "postgresql://user:pass@localhost:5432/mydb",
    "run_admin_server": False,
    "executor_id": "replica-1",  # unique per replica
})

使用 Postgres 时，多个服务副本可以共享同一个数据库，实现分布式执行与恢复。完整的多副本演示见 examples/dbos/ 目录。

Python 客户端

关于以编程方式与 WorkflowServer 交互，请看 Python 客户端文档。

服务端的 JSON 解码使用工作流声明的事件与 additional_events。请用 server.add_workflow(name, workflow, additional_events=[MyEvent]) 添加 API 事件。serializer= 设置只影响内部 context 与检查点。