Dapr Agents v1.0: Hướng Dẫn Xây Dựng AI Agent Workflows Production-Ready trên Dapr

from dapr.ext.workflow import WorkflowRuntime, DaprWorkflowContext

@workflow
def document_pipeline(ctx: DaprWorkflowContext, batch_id: str):
    # Mỗi bước được checkpoint tự động
    raw_docs = yield ctx.call_activity(extract_documents, input=batch_id)
    validated = yield ctx.call_activity(validate_batch, input=raw_docs)
    enriched = yield ctx.call_activity(enrich_with_llm, input=validated)
    return enriched

# Lưu agent state
await client.save_state("statestore", "agent-session-123", {
    "current_step": "enrichment",
    "processed_count": 47,
    "checkpoint_time": "2026-03-25T10:30:00Z"
})

# Đọc lại khi cần
state = await client.get_state("statestore", "agent-session-123")

# resiliency.yaml
apiVersion: dapr.io/v1alpha1
kind: Resiliency
metadata:
  name: llm-resiliency
spec:
  policies:
    retries:
      llm-retry:
        policy: exponential
        maxInterval: 15s
        maxRetries: 5
    timeouts:
      llm-timeout:
        duration: 30s
  targets:
    components:
      openai-binding:
        retry: llm-retry
        timeout: llm-timeout

# Xem traces trên Zipkin/Jaeger
# Metrics trên Prometheus
# Logs structured JSON format

# Sample trace output
{
  "traceId": "abc123",
  "spanId": "def456",
  "operationName": "agent.document_pipeline",
  "duration": "2.3s",
  "status": "SUCCESS",
  "attributes": {
    "agent.name": "document-processor",
    "workflow.step": "enrich_with_llm",
    "llm.provider": "openai",
    "tokens.total": 1847
  }
}

┌────────────────────────────────────────────────────────┐
│                   Client / Trigger Layer                │
│        (API Gateway / Event Bus / Scheduler)            │
└──────────────────────┬─────────────────────────────────┘
                       │
┌──────────────────────▼─────────────────────────────────┐
│              Orchestrator Agent                         │
│   - Nhận task từ trigger                               │
│   - Phân chia workflow steps                           │
│   - Quản lý sub-agents                                 │
└────────┬──────────────┬───────────────┬────────────────┘
         │              │               │
   ┌─────▼─────┐  ┌─────▼─────┐  ┌─────▼─────┐
   │ Extract   │  │ Validate  │  │  Enrich   │
   │  Agent    │  │  Agent    │  │  Agent    │
   └─────┬─────┘  └─────┬─────┘  └─────┬─────┘
         │              │               │
┌────────▼──────────────▼───────────────▼────────────────┐
│                   Dapr Sidecar Layer                    │
│  State Store │ Pub/Sub │ Bindings │ Secrets │ Resiliency │
└────────────────────────────────────────────────────────┘
         │              │               │
   ┌─────▼─────┐  ┌─────▼─────┐  ┌─────▼─────┐
   │  Redis /  │  │  OpenAI / │  │  Output   │
   │ Postgres  │  │  Anthropic│  │   Store   │
   └───────────┘  └───────────┘  └───────────┘

# Install Dapr CLI
wget -q https://raw.githubusercontent.com/dapr/cli/master/install/install.sh -O - | /bin/bash

# Init Dapr (local development)
dapr init

# Install Python SDK
pip install dapr-agents

# agents/pipeline.py
from dapr_agents import DaprAgent, workflow, activity
from dapr.ext.workflow import DaprWorkflowContext

@activity
async def extract_documents(batch_id: str) -> list[dict]:
    """Trích xuất documents từ storage"""
    # Logic đọc file/database
    docs = await storage.read_batch(batch_id)
    return [{"id": doc.id, "content": doc.text} for doc in docs]

@activity
async def validate_batch(docs: list[dict]) -> list[dict]:
    """Validate document format và content"""
    valid = []
    for doc in docs:
        if len(doc["content"]) > 100:  # minimum content check
            valid.append(doc)
    return valid

@activity
async def enrich_with_llm(docs: list[dict]) -> list[dict]:
    """Summarize và extract entities với LLM"""
    enriched = []
    for doc in docs:
        summary = await llm.summarize(doc["content"])
        entities = await llm.extract_entities(doc["content"])
        enriched.append({**doc, "summary": summary, "entities": entities})
    return enriched

@workflow
async def document_pipeline(ctx: DaprWorkflowContext, batch_id: str):
    # Mỗi yield = một checkpoint
    docs = yield ctx.call_activity(extract_documents, input=batch_id)
    valid_docs = yield ctx.call_activity(validate_batch, input=docs)
    result = yield ctx.call_activity(enrich_with_llm, input=valid_docs)
    return {"batch_id": batch_id, "processed": len(result), "data": result}

# Lưu checkpoint sau mỗi batch
@activity
async def save_checkpoint(ctx: dict) -> None:
    await dapr_client.save_state(
        store_name="statestore",
        key=f"pipeline-{ctx['batch_id']}-checkpoint",
        value=ctx
    )

@workflow
async def document_pipeline_resilient(ctx: DaprWorkflowContext, batch_id: str):
    try:
        docs = yield ctx.call_activity(
            extract_documents,
            input=batch_id,
            retry_policy=RetryPolicy(
                max_number_of_attempts=3,
                backoff_coefficients=2.0,
                initial_retry_internal=timedelta(seconds=5)
            )
        )
    except TaskFailedError as e:
        # Log và mark batch as failed
        yield ctx.call_activity(mark_batch_failed, input={
            "batch_id": batch_id,
            "error": str(e)
        })
        return {"status": "FAILED", "reason": str(e)}
    
    # Continue pipeline...

# Clone quickstarts chính thức
git clone https://github.com/dapr/dapr-agents.git
cd dapr-agents/quickstarts

# Chạy quickstart đầu tiên
dapr run --app-id agent-quickstart -- python quickstarts/hello_agent/hello_agent.py