Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -337,91 +337,79 @@ Workflows are structured processes where LLM agents and tools collaborate in pre
This approach is particularly suitable for business-critical applications where you need both the intelligence of LLMs and the reliability of traditional software systems.

```python
import dapr.ext.workflow as wf
from dapr.ext.workflow import DaprWorkflowContext

from dapr_agents.llm.dapr import DaprChatClient
from dapr_agents.workflow.decorators import llm_activity

runtime = wf.WorkflowRuntime()
llm = DaprChatClient(component_name="openai")
import time

import dapr.ext.workflow as wf

@runtime.workflow(name="task_chain_workflow")
def task_chain_workflow(ctx: DaprWorkflowContext):
character = yield ctx.call_activity(get_character)
line = yield ctx.call_activity(get_line, input={"character": character})
return line

wfr = wf.WorkflowRuntime()

@runtime.activity(name="get_character")
@llm_activity(
prompt="Pick a random character from The Lord of the Rings. Respond with the name only.",
llm=llm,
)
def get_character(ctx) -> str:
pass
@wfr.workflow(name="support_workflow")
def support_workflow(ctx: wf.DaprWorkflowContext, request: dict) -> str:
triage_result = yield ctx.call_child_workflow(
workflow="agent_workflow",
input={"task": f"Assist with the following support request:\n\n{request}"},
app_id="triage-agent",
)
if triage_result:
print("Triage result:", triage_result.get("content", ""), flush=True)

recommendation = yield ctx.call_child_workflow(
workflow="agent_workflow",
input={"task": triage_result.get("content", "")},
app_id="expert-agent",
)
if recommendation:
print("Recommendation:", recommendation.get("content", ""), flush=True)

@runtime.activity(name="get_line")
@llm_activity(
prompt="What is a famous line by {character}?",
llm=llm,
)
def get_line(ctx, character: str) -> str:
pass
return recommendation.get("content", "") if recommendation else ""

wfr.start()
time.sleep(5)

runtime.start()
client = wf.DaprWorkflowClient()
instance_id = client.schedule_new_workflow(task_chain_workflow)
state = client.wait_for_workflow_completion(instance_id)
print(state.serialized_output)
runtime.shutdown()
```

This workflow demonstrates sequential task execution where the output of one LLM-backed activity becomes the input for the next. The `@llm_activity` decorator wires prompts, formatting, and response handling so activities stay deterministic while still using model reasoning.

Dapr Agents supports coordination of LLM interactions at different levels of granularity:

### LLM Activities
`@llm_activity` binds a workflow activity to a prompt, LLM client, and optional structured output. The decorated function body can stay empty because the decorator handles prompting, retries, and response parsing.

```python
llm = DaprChatClient(component_name="openai")

@runtime.activity(name="generate_outline")
@llm_activity(
prompt="Create a short outline about {topic}.",
llm=llm,
request = {
"customer": "alice",
"issue": "Unable to access dashboard after recent update",
}
instance_id = client.schedule_new_workflow(
workflow=support_workflow,
input=request,
)
def generate_outline(ctx, topic: str) -> str:
pass
client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
wfr.shutdown()
```

LLM activities are perfect for lightweight reasoning steps, extraction tasks, or summarization stages that need deterministic workflow control with LLM flexibility.

### Agent Activities
`@agent_activity` lets workflows call fully-configured `Agent` instances (tools, memory, instructions) as activities. The workflow provides the inputs, and the decorator routes execution through the agent’s reasoning loop.
Here the `call_child_workflow` is used to invoke the workflow of two Dapr Agents and pass output from one as input to the other. This requires the `DurableAgent` to run as:

```python
planner = Agent(
name="PlannerAgent",
role="Trip planner",
instructions=["Create a concise 3-day plan for any city."],
llm=DaprChatClient(component_name="openai"),
)
from dapr_agents import DurableAgent
from dapr_agents.agents.configs import AgentMemoryConfig
from dapr_agents.llm.dapr import DaprChatClient
from dapr_agents.memory import ConversationDaprStateMemory
from dapr_agents.workflow.runners.agent import AgentRunner

@runtime.activity(name="plan_outline")
@agent_activity(agent=planner)
def plan_outline(ctx, destination: str) -> dict:
pass
expert_agent = DurableAgent(
name="expert_agent",
role="Technical Support Specialist",
goal="Provide recommendations based on customer context and issue.",
instructions=[
"Provide a clear, actionable recommendation to resolve the issue.",
],
llm=DaprChatClient(component_name="llm-provider"),
memory=AgentMemoryConfig(
store=ConversationDaprStateMemory(
store_name="agent-memory",
session_id=f"expert-agent-session",
)
),
)
runner = AgentRunner()
try:
runner.serve(expert_agent, port=8001)
finally:
runner.shutdown(expert_agent)
```

Agent activities enable workflows to leverage specialized agents with their own tools, memory, and reasoning capabilities while maintaining the structured coordination benefits of workflow orchestration.

> **Note:** Agent activities must use regular `Agent` instances, not `DurableAgent` instances, because workflows manage the execution context and durability through the Dapr workflow engine.

### Workflow Patterns

Workflows enable the implementation of various agentic patterns through structured orchestration, including Prompt Chaining, Routing, Parallelization, Orchestrator-Workers, Evaluator-Optimizer, Human-in-the-loop, and others. For detailed implementations and examples of these patterns, see the [Patterns documentation]({{< ref dapr-agents-patterns.md >}}).
Expand Down
Loading