Introduction
Dans mon expérience de cinq années en intégration de systèmes IA, j'ai confronté de nombreux défis autour de la gestion d'état dans les workflows Dify. La persistance des données utilisateur, la reprise sur erreur et la cohérence transactionnelle représentent des problématiques critiques que tout ingénieur doit maîtriser avant de déployer en production.
Ce tutoriel explore en profondeur l'architecture de persistence pour les workflows Dify, en utilisant une stratégie hybride PostgreSQL + Redis optimisée pour des latences inférieures à 50 millisecondes. Nous examinerons également comment
réduire vos coûts d'API de 85% avec HolySheep AI tout en maintenant des performances excelência.
Architecture de Persistence Multi-Niveaux
L'architecture de gestion d'état Dify repose sur trois couches distinctes : le cache Redis pour les accès chaud, la base PostgreSQL pour la persistance durable, et un bus d'événements pour la synchronisation inter-instances.
// Configuration Docker Compose pour Dify avec PostgreSQL et Redis
version: '3.8'
services:
postgres:
image: postgresql:15-alpine
environment:
POSTGRES_DB: dify_workflow
POSTGRES_USER: dify_admin
POSTGRES_PASSWORD: ${DB_PASSWORD}
volumes:
- postgres_data:/var/lib/postgresql/data
- ./init-schema.sql:/docker-entrypoint-initdb.d/init.sql
command: >
postgres
-c max_connections=200
-c shared_buffers=256MB
-c effective_cache_size=512MB
-c maintenance_work_mem=64MB
-c checkpoint_completion_target=0.9
-c wal_buffers=16MB
-c default_statistics_target=100
healthcheck:
test: ["CMD-SHELL", "pg_isready -U dify_admin"]
interval: 10s
timeout: 5s
retries: 5
redis:
image: redis:7-alpine
command: >
redis-server
--maxmemory 2gb
--maxmemory-policy allkeys-lru
--save 900 1
--save 300 10
--save 60 10000
--appendonly yes
--appendfsync everysec
volumes:
- redis_data:/data
sysctls:
net.core.somaxconn: 1024
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 10s
timeout: 5s
retries: 5
dify-api:
build:
context: .
dockerfile: Dockerfile.api
environment:
DB_HOST: postgres
DB_PORT: 5432
DB_USERNAME: dify_admin
DB_PASSWORD: ${DB_PASSWORD}
DB_DATABASE: dify_workflow
REDIS_HOST: redis
REDIS_PORT: 6379
REDIS_DB: 0
SECRET_KEY: ${SECRET_KEY}
CONSOLE_WEB_URL: "https://your-dify-instance.com"
depends_on:
postgres:
condition: service_healthy
redis:
condition: service_healthy
restart: unless-stopped
volumes:
postgres_data:
redis_data:
Schéma de Base de Données Optimisé
-- Migration SQL pour la gestion d'état workflow
-- Performance: Index composites sur (workflow_id, created_at) + Partitionnement par date
CREATE EXTENSION IF NOT EXISTS "uuid-ossp";
CREATE EXTENSION IF NOT EXISTS "pg_trgm";
-- Table principale des exécutions de workflow
CREATE TABLE workflow_executions (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
workflow_id UUID NOT NULL,
version INTEGER NOT NULL DEFAULT 1,
status VARCHAR(20) NOT NULL DEFAULT 'pending',
current_node_id VARCHAR(255),
input_data JSONB NOT NULL DEFAULT '{}',
output_data JSONB,
error_data JSONB,
execution_context JSONB NOT NULL DEFAULT '{}',
node_states JSONB NOT NULL DEFAULT '{}',
retry_count INTEGER NOT NULL DEFAULT 0,
max_retries INTEGER NOT NULL DEFAULT 3,
timeout_seconds INTEGER NOT NULL DEFAULT 3600,
started_at TIMESTAMPTZ,
completed_at TIMESTAMPTZ,
created_by UUID,
tenant_id UUID NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
CONSTRAINT valid_status CHECK (
status IN ('pending', 'running', 'completed', 'failed', 'cancelled', 'paused')
)
) PARTITION BY RANGE (created_at);
-- Index pour requêtes fréquentes
CREATE INDEX idx_workflow_exec_status ON workflow_executions(workflow_id, status);
CREATE INDEX idx_workflow_exec_tenant ON workflow_executions(tenant_id, created_at DESC);
CREATE INDEX idx_workflow_exec_current ON workflow_executions(status, updated_at)
WHERE status IN ('running', 'pending', 'paused');
-- Table de checkpoint pour reprise rapide
CREATE TABLE workflow_checkpoints (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
execution_id UUID NOT NULL REFERENCES workflow_executions(id) ON DELETE CASCADE,
checkpoint_key VARCHAR(255) NOT NULL,
state_data JSONB NOT NULL,
node_id VARCHAR(255) NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
UNIQUE(execution_id, checkpoint_key)
);
CREATE INDEX idx_checkpoint_execution ON workflow_checkpoints(execution_id, created_at);
-- Table de variables de session pour persistance cross-workflow
CREATE TABLE workflow_sessions (
session_id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
user_id UUID NOT NULL,
workflow_id UUID NOT NULL,
session_data JSONB NOT NULL DEFAULT '{}',
expires_at TIMESTAMPTZ NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_session_user ON workflow_sessions(user_id, workflow_id);
CREATE INDEX idx_session_expires ON workflow_sessions(expires_at)
WHERE expires_at > NOW();
-- Partition automatique par mois
CREATE OR REPLACE FUNCTION create_monthly_partition()
RETURNS void AS $$
DECLARE
partition_date DATE;
partition_name TEXT;
start_date DATE;
end_date DATE;
BEGIN
FOR i IN 0..2 LOOP
partition_date := DATE_TRUNC('month', CURRENT_DATE + (i || ' months')::INTERVAL);
partition_name := 'workflow_executions_' || TO_CHAR(partition_date, 'YYYY_MM');
start_date := partition_date;
end_date := partition_date + INTERVAL '1 month';
EXECUTE format(
'CREATE TABLE IF NOT EXISTS %I PARTITION OF workflow_executions
FOR VALUES FROM (%L) TO (%L)',
partition_name, start_date, end_date
);
END LOOP;
END;
$$ LANGUAGE plpgsql;
Service Python de Gestion d'État
# services/workflow_state_manager.py
"""
Gestionnaire d'état pour workflows Dify avec persistance hybride PostgreSQL/Redis
Performance cible: <50ms latence moyenne, 99th percentile <200ms
"""
import asyncio
import json
import hashlib
from datetime import datetime, timedelta
from typing import Optional, Dict, Any, List
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from enum import Enum
import asyncpg
import redis.asyncio as redis
from redis.asyncio import Redis
from asyncpg import Pool, Connection
import structlog
logger = structlog.get_logger(__name__)
class WorkflowStatus(Enum):
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
CANCELLED = "cancelled"
PAUSED = "paused"
@dataclass
class WorkflowState:
execution_id: str
workflow_id: str
status: WorkflowStatus
current_node_id: Optional[str] = None
input_data: Dict[str, Any] = field(default_factory=dict)
output_data: Optional[Dict[str, Any]] = None
node_states: Dict[str, Dict[str, Any]] = field(default_factory=dict)
context: Dict[str, Any] = field(default_factory=dict)
retry_count: int = 0
created_at: datetime = field(default_factory=datetime.utcnow)
updated_at: datetime = field(default_factory=datetime.utcnow)
class WorkflowStateManager:
"""
Gestionnaire optimisé pour la persistance d'état workflow.
Stratégie write-through avec cache Redis et flush asynchrone PostgreSQL.
"""
# Configuration des TTL Redis
CACHE_TTL_SHORT = 300 # 5 minutes pour nœuds actifs
CACHE_TTL_MEDIUM = 3600 # 1 heure pour exécutions récentes
CACHE_TTL_LONG = 86400 # 24 heures pour contexte utilisateur
# Seuil de batching pour optimisation I/O
FLUSH_THRESHOLD_MS = 100
MAX_BATCH_SIZE = 50
def __init__(
self,
pg_pool: Pool,
redis_client: Redis,
dry_run: bool = False
):
self.pg_pool = pg_pool
self.redis = redis_client
self.dry_run = dry_run
self._pending_writes: Dict[str, Dict[str, Any]] = {}
self._flush_lock = asyncio.Lock()
self._metrics = {
"cache_hits": 0,
"cache_misses": 0,
"db_writes": 0,
"flush_count": 0
}
def _get_cache_key(self, execution_id: str, suffix: str = "state") -> str:
"""Génération de clé cache optimisée avec hash"""
return f"wf:state:{execution_id}:{suffix}"
def _get_compression_key(self, data: Dict[str, Any]) -> str:
"""Hash rapide pour invalidation cohérente"""
serialized = json.dumps(data, sort_keys=True, default=str)
return hashlib.md5(serialized.encode()).hexdigest()[:12]
async def get_state(self, execution_id: str) -> Optional[WorkflowState]:
"""
Récupération d'état avec stratégie cache-aside.
Lecture Redis en priorité, fallback PostgreSQL.
"""
cache_key = self._get_cache_key(execution_id)
# Tentative lecture cache (durée <5ms)
cached = await self.redis.get(cache_key)
if cached:
self._metrics["cache_hits"] += 1
data = json.loads(cached)
return self._deserialize_state(data)
self._metrics["cache_misses"] += 1
# Fallback PostgreSQL
async with self.pg_pool.acquire() as conn:
row = await conn.fetchrow(
"""
SELECT id, workflow_id, status, current_node_id,
input_data, output_data, node_states, execution_context,
retry_count, created_at, updated_at
FROM workflow_executions
WHERE id = $1
""",
execution_id
)
if not row:
return None
state = self._row_to_state(row)
# Warm-up cache pour accès ultérieur
await self._cache_state(state)
return state
async def create_execution(
self,
workflow_id: str,
input_data: Dict[str, Any],
tenant_id: str,
created_by: Optional[str] = None,
max_retries: int = 3,
timeout_seconds: int = 3600
) -> WorkflowState:
"""Création d'une nouvelle exécution workflow."""
execution_id = str(uuid.uuid4())
now = datetime.utcnow()
state = WorkflowState(
execution_id=execution_id,
workflow_id=workflow_id,
status=WorkflowStatus.PENDING,
input_data=input_data,
created_at=now,
updated_at=now
)
# Écriture synchrone PostgreSQL pour durabilité immédiate
async with self.pg_pool.acquire() as conn:
await conn.execute(
"""
INSERT INTO workflow_executions
(id, workflow_id, status, input_data, execution_context,
retry_count, max_retries, timeout_seconds, tenant_id, created_by,
created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
""",
execution_id,
workflow_id,
state.status.value,
json.dumps(input_data),
json.dumps(state.context),
state.retry_count,
max_retries,
timeout_seconds,
tenant_id,
created_by,
now,
now
)
self._metrics["db_writes"] += 1
# Cache de l'état créé
await self._cache_state(state)
logger.info(
"execution_created",
execution_id=execution_id,
workflow_id=workflow_id,
tenant_id=tenant_id
)
return state
async def update_node_state(
self,
execution_id: str,
node_id: str,
node_data: Dict[str, Any],
is_terminal: bool = False
) -> None:
"""
Mise à jour incrémentale de l'état d'un nœud.
Écriture optimisée : cache immédiat + batch PostgreSQL.
"""
cache_key = self._get_cache_key(execution_id, f"node:{node_id}")
# Update cache immédiatement (<5ms)
await self.redis.setex(
cache_key,
self.CACHE_TTL_SHORT,
json.dumps({
"data": node_data,
"updated_at": datetime.utcnow().isoformat(),
"is_terminal": is_terminal
})
)
# Marquage pour flush asynchrone
async with self._flush_lock:
if execution_id not in self._pending_writes:
self._pending_writes[execution_id] = {
"nodes": {},
"updated_at": datetime.utcnow()
}
self._pending_writes[execution_id]["nodes"][node_id] = {
"data": node_data,
"is_terminal": is_terminal,
"timestamp": datetime.utcnow()
}
async def flush_pending_updates(self) -> int:
"""
Flush asynchrone des mises à jour en attente.
Batch optimal pour réduire les écritures I/O.
"""
async with self._flush_lock:
if not self._pending_writes:
return 0
executions_to_flush = list(self._pending_writes.keys())[:self.MAX_BATCH_SIZE]
flush_count = 0
async with self.pg_pool.acquire() as conn:
async with conn.transaction():
for execution_id in executions_to_flush:
updates = self._pending_writes.pop(execution_id)
await conn.execute(
"""
UPDATE workflow_executions
SET node_states = node_states || $2::jsonb,
updated_at = $3
WHERE id = $1
""",
execution_id,
json.dumps(updates["nodes"]),
datetime.utcnow()
)
flush_count += 1
self._metrics["db_writes"] += flush_count
self._metrics["flush_count"] += 1
logger.debug(
"flush_completed",
flush_count=flush_count,
pending_remaining=len(self._pending_writes)
)
return flush_count
async def create_checkpoint(
self,
execution_id: str,
checkpoint_key: str,
state_data: Dict[str, Any],
node_id: str
) -> str:
"""Création de point de restauration pour reprise rapide."""
checkpoint_id = str(uuid.uuid4())
async with self.pg_pool.acquire() as conn:
await conn.execute(
"""
INSERT INTO workflow_checkpoints
(id, execution_id, checkpoint_key, state_data, node_id)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (execution_id, checkpoint_key)
DO UPDATE SET state_data = $4, created_at = NOW()
""",
checkpoint_id,
execution_id,
checkpoint_key,
json.dumps(state_data),
node_id
)
# Mise à jour cache avec nouveau checkpoint
checkpoint_cache_key = self._get_cache_key(execution_id, f"checkpoint:{checkpoint_key}")
await self.redis.setex(
checkpoint_cache_key,
self.CACHE_TTL_LONG,
json.dumps(state_data)
)
return checkpoint_id
async def get_checkpoint(
self,
execution_id: str,
checkpoint_key: str
) -> Optional[Dict[str, Any]]:
"""Récupération de checkpoint avec fallback cache."""
checkpoint_cache_key = self._get_cache_key(execution_id, f"checkpoint:{checkpoint_key}")
cached = await self.redis.get(checkpoint_cache_key)
if cached:
return json.loads(cached)
async with self.pg_pool.acquire() as conn:
row = await conn.fetchrow(
"""
SELECT state_data FROM workflow_checkpoints
WHERE execution_id = $1 AND checkpoint_key = $2
""",
execution_id,
checkpoint_key
)
if row:
await self.redis.setex(
checkpoint_cache_key,
self.CACHE_TTL_LONG,
json.dumps(row['state_data'])
)
return row['state_data']
return None
@asynccontextmanager
async def transaction(self, execution_id: str):
"""Contexte transactionnel pour opérations atomiques."""
async with self.pg_pool.acquire() as conn:
async with conn.transaction():
yield TransactionContext(conn, execution_id, self)
async def _cache_state(self, state: WorkflowState) -> None:
"""Cache l'état complet pour lecture rapide."""
cache_key = self._get_cache_key(state.execution_id)
ttl = self.CACHE_TTL_MEDIUM if state.status in [
WorkflowStatus.COMPLETED, WorkflowStatus.FAILED
] else self.CACHE_TTL_SHORT
await self.redis.setex(
cache_key,
ttl,
json.dumps(self._serialize_state(state), default=str)
)
def _serialize_state(self, state: WorkflowState) -> Dict[str, Any]:
"""Sérialisation optimisée pour Redis."""
return {
"execution_id": state.execution_id,
"workflow_id": state.workflow_id,
"status": state.status.value,
"current_node_id": state.current_node_id,
"node_states": state.node_states,
"retry_count": state.retry_count,
"created_at": state.created_at.isoformat(),
"updated_at": state.updated_at.isoformat()
}
def _deserialize_state(self, data: Dict[str, Any]) -> WorkflowState:
"""Désérialisation depuis cache Redis."""
return WorkflowState(
execution_id=data["execution_id"],
workflow_id=data["workflow_id"],
status=WorkflowStatus(data["status"]),
current_node_id=data.get("current_node_id"),
node_states=data.get("node_states", {}),
retry_count=data.get("retry_count", 0),
created_at=datetime.fromisoformat(data["created_at"]),
updated_at=datetime.fromisoformat(data["updated_at"])
)
def _row_to_state(self, row) -> WorkflowState:
"""Conversion de ligne PostgreSQL vers objet Python."""
return WorkflowState(
execution_id=str(row["id"]),
workflow_id=str(row["workflow_id"]),
status=WorkflowStatus(row["status"]),
current_node_id=row.get("current_node_id"),
input_data=row.get("input_data", {}),
output_data=row.get("output_data"),
node_states=row.get("node_states", {}),
context=row.get("execution_context", {}),
retry_count=row.get("retry_count", 0),
created_at=row.get("created_at", datetime.utcnow()),
updated_at=row.get("updated_at", datetime.utcnow())
)
def get_metrics(self) -> Dict[str, Any]:
"""Métriques de performance pour monitoring."""
total_cache = self._metrics["cache_hits"] + self._metrics["cache_misses"]
cache_hit_rate = (
self._metrics["cache_hits"] / total_cache
if total_cache > 0 else 0
)
return {
**self._metrics,
"cache_hit_rate": round(cache_hit_rate, 4),
"pending_writes": len(self._pending_writes)
}
Intégration avec API LLM HolySheep
Dans mon utilisation quotidienne, j'ai constaté que
HolySheep AI offre des latences sub-50ms qui s'intègrent parfaitement avec notre architecture de workflow. Leur support WeChat et Alipay simplifie considérablement la gestion des paiements pour les équipes chinoises.
# services/llm_integration.py
"""
Intégration optimisée avec l'API HolySheep AI pour génération de contexte.
Tarification 2026 compétitive: DeepSeek V3.2 à $0.42/MTok soit 85%+ d'économie.
"""
import asyncio
import httpx
from typing import AsyncIterator, Dict, Any, Optional
from dataclasses import dataclass
import time
import structlog
from .workflow_state_manager import WorkflowStateManager, WorkflowState
logger = structlog.get_logger(__name__)
@dataclass
class LLMResponse:
content: str
model: str
usage: Dict[str, int]
latency_ms: float
cost_usd: float
class HolySheepLLMClient:
"""
Client HTTP optimisé pour HolySheep API.
Caractéristiques: retry exponentiel, circuit breaker, streaming prefetch.
"""
# Configuration API HolySheep
BASE_URL = "https://api.holysheep.ai/v1"
# Modèles disponibles avec tarification 2026
MODEL_PRICING = {
"gpt-4.1": {"input": 8.0, "output": 24.0}, # $/M tokens
"claude-sonnet-4.5": {"input": 15.0, "output": 75.0},
"gemini-2.5-flash": {"input": 2.50, "output": 10.0},
"deepseek-v3.2": {"input": 0.42, "output": 1.68},
}
# Configuration connexion
TIMEOUT_TOTAL = 60.0
TIMEOUT_CONNECT = 5.0
MAX_RETRIES = 3
RETRY_BASE_DELAY = 1.0
def __init__(
self,
api_key: str,
state_manager: WorkflowStateManager,
default_model: str = "deepseek-v3.2",
enable_caching: bool = True
):
self.api_key = api_key
self.state_manager = state_manager
self.default_model = default_model
self.enable_caching = enable_caching
# Configuration client HTTP optimisé
limits = httpx.Limits(
max_keepalive_connections=20,
max_connections=100,
keepalive_expiry=300
)
self._client = httpx.AsyncClient(
base_url=self.BASE_URL,
timeout=httpx.Timeout(
connect=self.TIMEOUT_CONNECT,
read=self.TIMEOUT_TOTAL,
write=30.0,
pool=10.0
),
limits=limits,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"X-Request-Timeout": str(int(self.TIMEOUT_TOTAL * 1000))
}
)
self._metrics = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"total_tokens": 0,
"total_cost_usd": 0.0,
"avg_latency_ms": 0.0
}
async def generate_with_context(
self,
execution_id: str,
prompt: str,
system_prompt: Optional[str] = None,
model: Optional[str] = None,
temperature: float = 0.7,
max_tokens: int = 2048,
**kwargs
) -> LLMResponse:
"""
Génération avec injection automatique du contexte workflow.
Met à jour l'état après chaque appel pour traçabilité.
"""
start_time = time.perf_counter()
model = model or self.default_model
# Récupération contexte workflow
state = await self.state_manager.get_state(execution_id)
if not state:
raise ValueError(f"Execution {execution_id} not found")
# Construction du prompt enrichi
enriched_prompt = self._build_enriched_prompt(
prompt=prompt,
state=state,
system_prompt=system_prompt
)
# Préparation messages
messages = []
if system_prompt or state.context.get("system_instructions"):
messages.append({
"role": "system",
"content": system_prompt or state.context.get("system_instructions", "")
})
messages.append({"role": "user", "content": enriched_prompt})
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
**kwargs
}
# Log avant appel
logger.debug(
"llm_request_start",
execution_id=execution_id,
model=model,
prompt_length=len(enriched_prompt)
)
try:
response = await self._execute_with_retry(payload)
latency_ms = (time.perf_counter() - start_time) * 1000
result = LLMResponse(
content=response["choices"][0]["message"]["content"],
model=model,
usage=response.get("usage", {}),
latency_ms=latency_ms,
cost_usd=self._calculate_cost(model, response.get("usage", {}))
)
# Mise à jour état workflow avec résultat
await self._update_workflow_state(
execution_id, state, result, enriched_prompt
)
self._update_metrics(result)
logger.info(
"llm_request_success",
execution_id=execution_id,
model=model,
latency_ms=round(latency_ms, 2),
cost_usd=round(result.cost_usd, 6)
)
return result
except Exception as e:
self._metrics["failed_requests"] += 1
logger.error(
"llm_request_failed",
execution_id=execution_id,
model=model,
error=str(e)
)
raise
async def stream_generate(
self,
execution_id: str,
prompt: str,
model: Optional[str] = None,
**kwargs
) -> AsyncIterator[str]:
"""
Génération en streaming avec yield progressif.
Idéal pour interfaces utilisateur temps réel.
"""
model = model or self.default_model
state = await self.state_manager.get_state(execution_id)
enriched_prompt = self._build_enriched_prompt(prompt, state)
messages = [
{"role": "user", "content": enriched_prompt}
]
payload = {
"model": model,
"messages": messages,
"stream": True,
**kwargs
}
accumulated_content = ""
async with self._client.stream("POST", "/chat/completions", json=payload) as response:
response.raise_for_status()
async for line in response.aiter_lines():
if line.startswith("data: "):
data = line[6:]
if data == "[DONE]":
break
chunk = json.loads(data)
content = chunk["choices"][0]["delta"].get("content", "")
if content:
accumulated_content += content
yield content
# Mise à jour finale de l'état
await self.state_manager.update_node_state(
execution_id=execution_id,
node_id="llm_stream_node",
node_data={
"output": accumulated_content,
"model": model,
"streamed": True
}
)
async def _execute_with_retry(self, payload: Dict[str, Any]) -> Dict[str, Any]:
"""Exécution avec retry exponentiel et backoff jitter."""
last_exception = None
for attempt in range(self.MAX_RETRIES):
try:
response = await self._client.post(
"/chat/completions",
json=payload
)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
last_exception = e
if e.response.status_code in [429, 500, 502, 503, 504]:
delay = self.RETRY_BASE_DELAY * (2 ** attempt) + random.uniform(0, 1)
logger.warning(
"llm_retry",
attempt=attempt + 1,
delay=delay,
status=e.response.status_code
)
await asyncio.sleep(delay)
else:
raise
except httpx.RequestError as e:
last_exception = e
delay = self.RETRY_BASE_DELAY * (2 ** attempt)
await asyncio.sleep(delay)
raise last_exception
def _build_enriched_prompt(
self,
prompt: str,
state: WorkflowState,
system_prompt: Optional[str] = None
) -> str:
"""Enrichissement du prompt avec contexte workflow."""
context_parts = []
if state.node_states:
context_parts.append("## Historique d'exécution:\n")
for node_id, node_data in state.node_states.items():
if node_data.get("output"):
context_parts.append(
f"- **{node_id}**: {node_data['output'][:500]}"
)
if state.context.get("variables"):
context_parts.append("\n## Variables:\n")
for key, value in state.context["variables"].items():
context_parts.append(f"- {key}: {value}")
context_str = "\n".join(context_parts) if context_parts else "Aucun contexte précédent."
return f"""## Contexte Workflow
{context_str}
Requête actuelle
{prompt}"""
async def _update_workflow_state(
self,
execution_id: str,
state: WorkflowState,
result: LLMResponse,
prompt: str
) -> None:
"""Mise à jour de l'état workflow après génération."""
await self.state_manager.update_node_state(
execution_id=execution_id,
node_id="llm_generation",
node_data={
"prompt": prompt[:2000],
"output": result.content,
"model": result.model,
"usage": result.usage,
"latency_ms": result.latency_ms,
"cost_usd": result.cost_usd,
"timestamp": datetime.utcnow().isoformat()
},
is_terminal=False
)
def _calculate_cost(self, model: str, usage: Dict[str, int]) -> float:
"""Calcul précis du coût en dollars."""
pricing = self.MODEL_PRICING.get(model, self.MODEL_PRICING["deepseek-v3.2"])
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
return input_cost + output_cost
def _update_metrics(self, result: LLMResponse) -> None:
"""Mise à jour métriques agrégées."""
self._metrics["total_requests"] += 1
self._metrics["successful_requests"] += 1
self._metrics["total_tokens"] += (
result.usage.get("prompt_tokens", 0) +
result.usage.get("completion_tokens", 0)
)
self._metrics["total_cost_usd"] += result.cost_usd
# Moyenne mobile exponentielle pour latence
alpha = 0.1
self._metrics["avg_latency_ms"] = (
alpha * result.latency_ms +
(1 - alpha) * self._metrics["avg_latency_ms"]
)
def get_cost_summary(self) -> Dict[str, Any]:
"""Résumé des coûts pour reporting."""
return {
"total_requests": self._metrics["total_requests"],
"total_tokens": self._metrics["total_tokens"],
"total_cost_usd": round(self._metrics["total_cost_usd"], 6),
"avg_latency_ms": round(self._metrics["avg_latency_ms"], 2),
"cost_per_1k_tokens": round(
(self._metrics["total_cost_usd"] / self._metrics["total_tokens"] * 1000)
if self._metrics["total_tokens"] > 0 else 0,
6
)
}
async def close(self):
"""Fermeture propre des connexions."""
await self._client.aclose()
Exemple d'utilisation
async def example_workflow_with_holysheep():
"""Exemple complet d'intégration workflow + LLM."""
import uuid
# Configuration (remplacer par vos vraies valeurs)
pg_pool = await asyncpg.create_pool(
host="localhost",
port=5432,
user="dify_admin",
password="secure_password",
database="dify_workflow",
min_size=10,
max_size=20
)
redis_client = redis.from_url("redis://localhost:6379/0")
# Initialisation des services
state_manager = WorkflowStateManager(pg_pool, redis_client)
llm_client = HolySheepLLMClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
state_manager=state_manager,
default_model="deepseek-v3.2" # Modèle le plus économique
)
try:
# Création d'une exécution workflow
execution = await state_manager.create_execution(
workflow_id=str(uuid.uuid4()),
input_data={"user_query": "Explique la gestion d'état"},
tenant_id="tenant_123",
created_by="user_456"
)
# Génération avec contexte
response = await llm_client.generate_with_context(
execution_id=execution.execution_id,
prompt="Donne-moi les bonnes pratiques pour PostgreSQL",
system_prompt="Tu es un expert en bases de données.",
temperature=0.7
)
print(f"Réponse: {response.content[:200]}...")
print(f"Latence: {response.latency_ms:.2f}ms")
print(f"Coût: ${response.cost_usd:.6f}")
# Statistiques de coût
summary = llm_client.get_cost_summary()
print(f"Coût total: ${summary['total_cost_usd']}")
finally:
await llm_client.close()
await pg_pool.close()
await redis_client.close()