This is the multi-page printable view of this section. Click here to print.
Dapr 为 CrewAI 代理提供一流集成，涵盖代理会话管理、通过发布订阅连接代理以及编排代理工作流。
Dapr Workflows 使 CrewAI 代理能够可靠地、持久地且具备内置弹性地运行。
通过 Dapr Workflow 引擎编排 CrewAI 任务，开发者可以：
本指南将介绍如何使用 Dapr Workflows 编排多个 CrewAI 任务，确保即使进程重启，每个步骤也只执行一次。
在本地初始化 Dapr 以设置自托管开发环境。此过程会安装 Dapr 边车二进制文件、配置工作流引擎并准备默认组件目录。有关完整详情，请参阅本地初始化 Dapr 的指南。
初始化 Dapr：
dapr init
验证 daprio/dapr、openzipkin/zipkin 和 redis 是否正在运行：
docker ps
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
pip install dapr dapr-ext-workflow crewai
创建一个名为 crewai_workflow.py 的文件并粘贴以下内容：
from dapr.ext.workflow import (
WorkflowRuntime,
DaprWorkflowContext,
WorkflowActivityContext,
DaprWorkflowClient,
)
from crewai import Agent, Task, Crew
import time
wfr = WorkflowRuntime()
# ------------------------------------------------------------
# 1. 定义代理、任务和任务字典
# ------------------------------------------------------------
agent = Agent(
role="Research Analyst",
goal="Research and summarize impactful technology updates.",
backstory="A skilled analyst who specializes in researching and summarizing technology topics.",
)
tasks = {
"latest_ai_news": Task(
description="Find the latest news about artificial intelligence.",
expected_output="A 3-paragraph summary of the top 3 stories.",
agent=agent,
),
"ai_startup_launches": Task(
description="Summarize the most impactful AI startup launches in the last 6 months.",
expected_output="A list summarizing 2 AI startups with links.",
agent=agent,
),
"ai_policy_updates": Task(
description="Summarize the newest AI government policy and regulation updates.",
expected_output="A bullet-point list summarizing the latest policy changes.",
agent=agent,
),
}
# ------------------------------------------------------------
# 2. 活动 — 按名称运行一个任务
# ------------------------------------------------------------
@wfr.activity(name="run_task")
def run_task_activity(ctx: WorkflowActivityContext, task_name: str):
print(f"Running CrewAI task: {task_name}", flush=True)
task = tasks[task_name]
# 为这单个任务创建一个 Crew
temp_crew = Crew(agents=[agent], tasks=[task])
# kickoff() 适用于所有 CrewAI 版本
result = temp_crew.kickoff()
return str(result)
# ------------------------------------------------------------
# 3. 工作流 — 持久地编排任务
# ------------------------------------------------------------
@wfr.workflow(name="crewai_multi_task_workflow")
def crewai_workflow(ctx: DaprWorkflowContext):
print("Starting multi-task CrewAI workflow", flush=True)
latest_news = yield ctx.call_activity(run_task_activity, input="latest_ai_news")
startup_summary = yield ctx.call_activity(run_task_activity, input="ai_startup_launches")
policy_updates = yield ctx.call_activity(run_task_activity, input="ai_policy_updates")
return {
"latest_news": latest_news,
"startup_summary": startup_summary,
"policy_updates": policy_updates,
}
# ------------------------------------------------------------
# 4. 运行时 + 客户端（入口点）
# ------------------------------------------------------------
if __name__ == "__main__":
wfr.start()
client = DaprWorkflowClient()
instance_id = "crewai-multi-01"
client.schedule_new_workflow(
workflow=crewai_workflow,
input=None,
instance_id=instance_id
)
state = client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
print(state.serialized_output)
这个 CrewAI 代理启动了一个工作流，用于收集和汇总 AI 和创业公司相关新闻。
Dapr Workflows 使用任何支持工作流的 Dapr 状态存储 来持久化状态。
创建一个名为 components 的目录，然后创建文件 workflowstore.yaml：
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
name: workflowstore
spec:
type: state.redis
version: v1
metadata:
- name: redisHost
value: localhost:6379
- name: redisPassword
value: ""
- name: actorStateStore
value: "true"
此组件存储：
CrewAI 需要 LLM 配置或令牌才能运行。请参阅此处的说明。
例如，设置 OpenAI：
export OPENAI_API_KEY=sk-...
使用 Dapr CLI 启动 CrewAI 工作流：
dapr run \
--app-id crewaiwf \
--dapr-grpc-port 50001 \
--resources-path ./components \
-- python3 ./crewai_workflow.py
工作流运行时，每个 CrewAI 任务都作为持久活动执行。 如果进程崩溃，工作流会从中断的地方精确恢复。您可以通过在第一个活动后终止进程，然后使用相同的应用 ID 重新运行上述命令来尝试此操作。
打开 Zipkin 查看工作流追踪：
http://localhost:9411
Dapr 为 LangGraph agents 提供了与 agent 会话管理（checkpointers）的一等集成。
Dapr Python SDK 通过 dapr-ext-langgraph 扩展与 LangGraph 检查点器集成。
在本地初始化 Dapr 以设置用于开发的自托管环境。此过程会获取并安装 Dapr 边车二进制文件，将基本服务作为 Docker 容器运行，并为您的应用程序准备默认的组件文件夹。有关详细步骤，请参阅官方本地初始化 Dapr 指南。
要初始化 Dapr 控制平面容器并创建默认配置文件，请运行：
dapr init
使用 docker ps 验证您是否有运行 daprio/dapr、openzipkin/zipkin 和 redis 镜像的容器实例：
docker ps
使用以下命令下载并安装 Dapr LangGraph 扩展：
pip install dapr-ext-langgraph langchain_openai langchain_core langgraph langgraph-prebuilt
开发包将包含与 Dapr 运行时预发布版本兼容的功能和行为。在安装 <code>dapr-dev</code> 包之前，请确保已卸载任何稳定版本的 Python SDK 扩展。
pip install dapr-ext-langgraph-dev langchain_openai langchain_core langgraph langgraph-prebuilt
要让 Dapr 处理 agent 的内存，请在编译图时使用 DaprCheckpointer 作为检查点器对象。像传递其他检查点器提供程序一样传递检查点器：
from dapr.ext.langgraph import DaprCheckpointer
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition
def add(a: int, b: int) -> int:
"""Adds a and b.
Args:
a: first int
b: second int
"""
return a + b
tools = [add]
llm = ChatOpenAI(model="gpt-4o")
llm_with_tools = llm.bind_tools(tools)
sys_msg = SystemMessage(
content='You are a helpful assistant tasked with performing arithmetic on a set of inputs.'
)
def assistant(state: MessagesState):
return {'messages': [llm_with_tools.invoke([sys_msg] + state['messages'])]}
builder = StateGraph(MessagesState)
builder.add_node('assistant', assistant)
builder.add_node('tools', ToolNode(tools))
builder.add_edge(START, 'assistant')
builder.add_conditional_edges(
'assistant',
tools_condition,
)
builder.add_edge('tools', 'assistant')
memory = DaprCheckpointer(store_name='statestore', key_prefix='dapr')
react_graph_memory = builder.compile(checkpointer=memory)
config = {'configurable': {'thread_id': '1'}}
messages = [HumanMessage(content='Add 3 and 4.')]
messages = react_graph_memory.invoke({'messages': messages}, config)
for m in messages['messages']:
m.pretty_print()
export OPENAI_API_KEY=sk-...
python -m venv .venv
source .venv/bin/activate # 在 Windows 上: .venv\Scripts\activate
组件文件是 Dapr 连接到数据库的方式。支持的数据库完整列表可以在这里找到。创建一个 components 目录并在其中创建此文件：
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
name: statestore
spec:
type: state.redis
version: v1
metadata:
- name: redisHost
value: localhost:6379
- name: redisPassword
value: ""
现在您已经有了一个使用 Dapr 管理 agent 会话的 LangGraph agent，进一步探索您可以使用 State API 做更多事情，以及如何启用弹性策略以增强可靠性。
Dapr 提供了 OpenAI agents 的一等集成，范围涵盖从 agent 会话管理到通过发布订阅连接 agents 以及编排 agentic 工作流。Dapr OpenAI 集成是 OpenAI Python SDK 中的一个扩展，开发者可以使用它通过各种 Dapr API 来增强 OpenAI agents。
通过使用 Dapr 管理 OpenAI Agent 的状态和会话数据，用户可以将 Agent 状态存储在 Dapr 支持的所有数据库中，包括键值存储、缓存和 SQL 数据库。开发者还可以获得内置的跟踪、指标和弹性策略，使 Agent 会话数据在生产环境中可靠运行。
在本地初始化 Dapr 以设置自托管开发环境。此过程获取并安装 Dapr 边车二进制文件，将必要的服务作为 Docker 容器运行，并准备应用程序的默认组件文件夹。详细步骤，请参阅官方本地初始化 Dapr 指南。
要初始化 Dapr 控制平面容器并创建默认配置文件，请运行：
dapr init
验证你正在运行的容器实例是否包含 daprio/dapr、openzipkin/zipkin 和 redis 镜像：
docker ps
pip install openai-agents dapr
让我们创建一个简单的 OpenAI Agent。将以下代码放入名为 openai_agent.py 的文件中：
import asyncio
from agents import Agent, Runner
from agents.extensions.memory.dapr_session import DaprSession
async def main():
agent = Agent(
name="Assistant",
instructions="Reply very concisely.",
)
session = DaprSession.from_address(
session_id="123",
state_store_name="statestore"
)
result = await Runner.run(agent, "What city is the Golden Gate Bridge in?", session=session)
print(result.final_output)
result = await Runner.run(agent, "What state is it in?", session=session)
print(result.final_output)
result = await Runner.run(agent, "What's the population?", session=session)
print(result.final_output)
asyncio.run(main())
export OPENAI_API_KEY=sk-...
python -m venv .venv
source .venv/bin/activate # 在 Windows 上：.venv\Scripts\activate
组件文件是 Dapr 连接数据库的方式。完整的支持数据库列表可以在这里找到这里。创建一个 components 目录并在其中创建此文件：
statestore.yaml：
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
name: statestore
spec:
type: state.redis
version: v1
metadata:
- name: redisHost
value: localhost:6379
- name: redisPassword
value: ""
现在使用 Dapr CLI 运行本地 Dapr 进程和 Python 脚本。
dapr run --app-id openaisessions --dapr-grpc-port 50001 --resources-path ./components -- python3 ./openai_agent.py
打开 http://localhost:9411 查看追踪和依赖关系图。
你可以使用以下命令查看存储在 Redis 中的会话数据
hgetall "123:messages"
现在你已拥有使用 Dapr 管理 Agent 会话的 OpenAI Agent，可以进一步探索状态 API的更多功能，以及如何启用弹性策略以增强可靠性。
在此处了解更多关于 OpenAI Agent 会话和 Dapr 的信息详情。