juan/forja
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
b4964261eca334778177c6ba2901cdebd0b6b755
forja/docs/componentes.md
T
juan b4964261ec proyecto generalizado
2026-05-23 16:44:45 +02:00

29 KiB
Raw Blame History

Componentes de Forja y su interrelación (bajo nivel)

Alcance. Este documento es la referencia de cableado: módulos exactos, firmas, el grafo de dependencias de imports, el grafo de inyección de dependencias, los contratos entre capas y las cadenas de llamada de cada endpoint. Es preciso, no narrativo.

Rutas relativas a core/src/ forja_core/ salvo que se diga otra cosa. Refleja el estado del repo en v0.1.0.

1. Grafo de dependencias de módulos (imports internos)

Edges = "X importa de Y". El grafo es un DAG; las capas de abajo no importan nada de las de arriba. Nivel = profundidad topológica.

NIVEL 0  (no importan nada del proyecto)
   config                        ← Settings (pydantic-settings, lee .env)
   observability.logging         ← configure_logging / bind_trace_id / clear_trace_id
   domain.agent                  ← AgentDefinition, LLMConfig, AgentVersionMeta
   domain.guardrail              ← GuardrailViolation
   domain.policy                 ← PolicyDefinition, PolicyValidator, PolicyVersionMeta
   registry.versioning           ← compute_hash, unified_diff, DiffResult
   llm.base                      ← LLMProvider (Protocol), Message, CompletionResult
   runtime.state                 ← AgentState (TypedDict)
   runtime.checkpointer          ← build_checkpointer (AsyncSqliteSaver)

NIVEL 1
   domain.execution              ⇐ domain.guardrail
   llm.mock / llm.azure / llm.openai   ⇐ llm.base
   registry.repository           ⇐ domain.agent, registry.versioning
   registry.policy_store         ⇐ domain.policy
   guardrails.validators         ⇐ domain.guardrail
   guardrails.base               ⇐ domain.guardrail, domain.policy
   api.persistence               ⇐ domain.execution, domain.guardrail
   api.middlewares               ⇐ observability.logging

NIVEL 2
   llm.factory                   ⇐ config, llm.base, llm.{mock,azure,openai}
   registry.factory              ⇐ config, registry.{repository,policy_store}
   guardrails.guardrails_ai      ⇐ domain.{guardrail,policy}, guardrails.validators
   guardrails.composite          ⇐ domain.{guardrail,policy}, guardrails.base
   guardrails.nemo               ⇐ domain.{guardrail,policy}

NIVEL 3
   guardrails.factory            ⇐ config, guardrails.{base,composite,guardrails_ai,nemo}
   runtime.nodes                 ⇐ domain.{agent,policy}, guardrails.base, llm.base, runtime.state
   runtime.graph                 ⇐ domain.{agent,policy}, guardrails.base, llm.base, runtime.{nodes,state}

NIVEL 4
   runtime.orchestrator          ⇐ domain.{agent,execution,guardrail,policy}, guardrails.base, llm.base, runtime.{checkpointer,graph}

NIVEL 5
   api.deps                      ⇐ config, guardrails.{base,factory}, llm.{base,factory}, registry.{factory,policy_store,repository}, runtime.orchestrator
   api.agents                    ⇐ api.deps, domain.agent, registry.versioning
   api.policies                  ⇐ api.deps, domain.policy
   api.violations                ⇐ api.deps, api.persistence, domain.guardrail
   api.executions                ⇐ api.deps, api.persistence, domain.{agent,execution,policy}, registry.{policy_store,repository}, runtime.orchestrator

NIVEL 6
   main                          ⇐ api (routers), api.middlewares, config, observability.logging   →  create_app(), app

(separado, sin imports del paquete core)
   dashboard/src/ forja_dashboard/*   ←  habla con `main` por HTTP, no por import

Reglas que se cumplen y conviene mantener:

  • domain/ no importa nada del resto del proyecto. Es el vocabulario; todos dependen de él.
  • Solo api/ importa de runtime.orchestrator (y deps.py lo construye). El resto de api/ no toca LangGraph; runtime/ no toca api/.
  • Solo los factory.py y main.py importan config.Settings. El resto recibe los objetos ya construidos.
  • guardrails.composite no conoce a sus sub-engines concretos (solo GuardrailEngine); los ensambla guardrails.factory.

2. Inyección de dependencias — api/deps.py

Cinco objetos del dominio, cada uno construido una sola vez por proceso (@lru_cache(maxsize=1)), más el orchestrator que los compone:

get_settings()  ──────────────────────────────────────────────►  Settings()      (pydantic-settings ← .env)
     │
     ├──► get_registry()        = build_agent_registry(settings)  ──►  FileSystemAgentRegistry(settings.agents_dir)
     ├──► get_policy_store()     = build_policy_store(settings)    ──►  FileSystemPolicyStore(settings.policies_dir)
     ├──► get_llm_provider()     = build_llm_provider(settings)    ──►  Mock | AzureOpenAI | OpenAI  (según settings.llm_provider)
     ├──► get_guardrail_engine() = build_guardrail_engine(settings)──►  CompositeGuardrailEngine([GuardrailsAIEngine(), (NeMoGuardrailsEngine() si settings.guardrails_nemo_enabled)])
     │
     └──► get_orchestrator()  =  AgentOrchestrator(
                                     provider = get_llm_provider(),
                                     engine   = get_guardrail_engine(),
                                     data_dir = get_settings().data_dir,
                                 )

Aliases que los routers piden por parámetro (Annotated[T, Depends(get_*)]):

Alias Tipo Lo usan
SettingsDep Settings executions (todos los endpoints), violations
RegistryDep FileSystemAgentRegistry agents (todos), executions (invoke, get, list, approve, reject)
PolicyStoreDep FileSystemPolicyStore policies (todos), executions (invoke, get, list, approve, reject)
OrchestratorDep AgentOrchestrator executions (invoke, get, list, approve, reject)

No hay LLMProviderDep ni GuardrailEngineDep: el provider y el engine solo se inyectan al AgentOrchestrator (vía get_orchestrator), nunca a un router.

Lifecycle: las get_* son perezosas → se llaman en la primera request que las necesita. Los tests hacen deps.get_settings.cache_clear() (etc.) tras monkeypatch.setenv("DATA_DIR", tmp_path) para reconstruir todo apuntando a un directorio temporal. main.create_app() llama a Settings() directamente (para configurar el logging), no a deps.get_settings(); son instancias distintas pero leen la misma config.

3. Contratos entre componentes (firmas exactas)

3.1 Proveedor LLM — llm/base.py

class Message(BaseModel):                 role: Literal["system","user","assistant"]; content: str
class CompletionResult(BaseModel):        content: str; model: str; tokens_in: int; tokens_out: int; latency_ms: int

class LLMProvider(Protocol):
    name: str
    async def complete(self, messages: list[Message], schema: dict[str,Any] | None = None,
                       temperature: float = 0.2, max_tokens: int = 2000) -> CompletionResult: ...

Implementaciones: MockProvider (llm/mock.py, determinista — elige una de _CANONICAL_RESPONSES = {"sip":…, "mos":…, "hss":…} buscando esas subcadenas en el input, en ese orden; si no, una respuesta genérica), AzureOpenAIProvider (llm/azure.py), OpenAIProvider (llm/openai.py). Selector: build_llm_provider(settings) en llm/factory.py (match settings.llm_provider). El nodo llm_reason lo invoca así: await provider.complete(messages=[Message("system", agent_def.system_prompt), Message("user", state["user_input"])], temperature=agent_def.llm.temperature, max_tokens=agent_def.llm.max_tokens).

3.2 Motor de guardrails — guardrails/base.py

class GuardrailEngine(Protocol):
    name: str
    async def validate_input (self, payload: str,            policy: PolicyDefinition, trace_id: UUID) -> list[GuardrailViolation]: ...
    async def validate_output(self, payload: dict[str, Any], policy: PolicyDefinition, trace_id: UUID) -> list[GuardrailViolation]: ...

class GuardrailViolation(BaseModel):
    trace_id: UUID; timestamp: datetime
    stage: Literal["input","output"]; validator: str
    severity: Literal["info","warning","block"]; message: str; blocked: bool

Implementaciones:

  • CompositeGuardrailEngine(engines: list[GuardrailEngine]) (guardrails/composite.py) — validate_* hace asyncio.gather(*(e.validate_*(...) for e in engines)) y aplana las listas. Lanza ValueError si engines está vacío.

  • GuardrailsAIEngine() (guardrails/guardrails_ai.py) — el real. Mantiene dos registros type → función:

    • INPUT_VALIDATORS = {"detect_pii", "prompt_injection", "toxic_language", "forbidden_topics"}
    • OUTPUT_VALIDATORS = {"schema_match", "pii_leakage", "forbidden_action_keywords", "telco_safety_rules"}

    _run(...) recorre policy.input_validators / policy.output_validators (cada uno un PolicyValidator(type, config)), busca type en el registro, y llama fn(payload, validator.config, trace_id, kind). Si fn lanza y policy.on_validator_error == "fail_closed" ⇒ añade una GuardrailViolation(severity="block", blocked=True, message=f"validator failed: {exc}", validator=v.type). Si el type no existe en el registro ⇒ log warning y continúa (no bloquea).

  • NeMoGuardrailsEngine(allowed_keywords) (guardrails/nemo.py) — stub; solo se añade al composite si settings.guardrails_nemo_enabled. validate_input: si el texto no menciona ningún keyword permitido ⇒ una violación severity="warning", blocked=False. validate_output: siempre [].

Selector: build_guardrail_engine(settings) en guardrails/factory.py.

Contrato de una función validadora (todas las de guardrails/validators.py):

def <validador>(payload: <str|dict>, config: dict[str, Any], trace_id: UUID, stage: str) -> list[GuardrailViolation]
  • Entrada (validate_input): payload es str (el texto del usuario). detect_pii, prompt_injection, toxic_language, forbidden_topics.
  • Salida (validate_output): payload es dict (el JSON parseado del LLM). schema_match, pii_leakage, forbidden_action_keywords, telco_safety_rules.
  • config viene literal del YAML de la política (p. ej. {entities:[...], severity_on_match:"block"}).
  • detect_pii: usa _presidio_analyzer() (singleton perezoso de presidio_analyzer.AnalyzerEngine con el modelo spaCy en_core_web_sm); si Presidio no está instalado o el modelo no carga, hace fallback a _pii_regex_fallback (regex de EMAIL / PHONE 3-3-3 / ES_NIF \d{8}[A-HJ-NP-TV-Z] / IP).

3.3 Registry de agentes — registry/repository.py → FileSystemAgentRegistry(root: Path)

Método Devuelve Lee
list_agents() list[AgentDefinition] recorre root/*/index.yaml, devuelve la versión activa de cada uno
get_agent(name, version=None) AgentDefinition index.yaml["active_version"] (o version) → get_version
get_version(name, version) AgentDefinition root/<name>/versions/<version>.yaml (lanza FileNotFoundError)
list_versions(name) list[AgentVersionMeta] index.yaml["versions"]
upsert_version(name, body, message, author) AgentVersionMeta escribe versions/<id>.yaml (con compute_hash) y actualiza index.yaml (si body.state=="active" cambia active_version)
diff_versions(name, v1, v2) DiffResult versioning.unified_diff entre los dos YAML

3.4 Store de políticas — registry/policy_store.py → FileSystemPolicyStore(root: Path)

list_policies() -> list[PolicyDefinition], get_policy(name, version=None) -> PolicyDefinition (lanza FileNotFoundError), list_versions(name) -> list[PolicyVersionMeta].

class PolicyValidator(BaseModel):     type: str; config: dict[str,Any] = {}
class PolicyDefinition(BaseModel):
    name: str; version: str; description: str
    input_validators:  list[PolicyValidator]
    output_validators: list[PolicyValidator]
    on_validator_error: Literal["fail_open","fail_closed"] = "fail_closed"
class PolicyVersionMeta(BaseModel):   id: str; hash: str; author: str; message: str; created_at: datetime

3.5 Orchestrator — runtime/orchestrator.py → AgentOrchestrator

AgentOrchestrator(*, provider: LLMProvider, engine: GuardrailEngine, data_dir: Path)

async invoke (*, agent_def: AgentDefinition, policy: PolicyDefinition, user_input: str,
              trace_id: UUID | None = None) -> AgentExecution
async resume (*, agent_def, policy, trace_id: UUID, decision: dict[str,Any]) -> AgentExecution
async snapshot(*, agent_def, policy, trace_id: UUID) -> AgentExecution | None
  • Cada uno abre su propio AsyncSqliteSaver (runtime/checkpointer.build_checkpointer(data_dir) → data_dir/checkpoints.sqlite, context manager async) y construye el grafo con build_graph(...).
  • invoke: arma el AgentState inicial (status="running", listas vacías…) y hace graph.ainvoke(state, config={"configurable": {"thread_id": str(trace_id)}}). Si el grafo lanza, marca crashed=True. Luego _snapshot(...) → AgentExecution.
  • resume: graph.ainvoke(Command(resume=decision), config={thread_id: trace_id}) — reanuda el interrupt() con decision.
  • snapshot: graph.aget_state(config); si state.values está vacío → None; si no, _build_execution(...).

3.6 El grafo — runtime/graph.py + runtime/nodes.py + runtime/state.py

# runtime/state.py
class AgentState(TypedDict, total=False):
    trace_id: str; agent_name: str; agent_version: str; user_input: str
    messages: list[dict]; raw_llm_output: str | None; parsed_output: dict | None
    proposed_actions: list[dict]; violations: list[dict]
    decision_path: Annotated[list[dict], operator.add]   # ← reducer: cada nodo AÑADE pasos
    status: str; error: str | None
    human_decision: dict | None; final_output: dict | None

# runtime/nodes.py
NodeFn = Callable[[AgentState], Awaitable[dict[str, Any]]]
# factories que devuelven un NodeFn, parametrizadas con engine/policy/provider/agent_def:
build_node_validate_input(engine, policy)        build_node_llm_reason(provider, agent_def)
build_node_validate_output(engine, policy)       build_node_propose_actions()
build_node_approve_gate(agent_def)               build_node_finalize()
# cada nodo añade un DecisionStep a decision_path: {step, timestamp, duration_ms, detail}

# runtime/graph.py
build_graph(*, agent_def, policy, provider, engine, checkpointer: BaseCheckpointSaver) -> CompiledGraph

Cableado (nodos y aristas):

START ─► validate_input
validate_input ──(conditional)──►  END   si status == "blocked_by_guardrail"
                                   llm_reason  en otro caso
llm_reason ──(conditional)──────►  END   si status == "failed"
                                   validate_output  en otro caso
validate_output ──(conditional)─►  END   si status in {"blocked_by_guardrail","failed"}
                                   propose_actions  en otro caso
propose_actions ──► approve_gate ──► finalize ──► END        (aristas fijas)
  • approve_gate: si hay acciones con risk_score >= agent_def.risk_threshold_for_hitl o requires_approval ⇒ decision = interrupt({"awaiting_actions": risky}) → el grafo se pausa ahí (LangGraph persiste el estado en el checkpointer); al reanudar con Command(resume=decision), interrupt() devuelve decision y el nodo continúa, escribiendo human_decision.
  • finalize: si human_decision.rejected ⇒ status="failed", error="rejected_by_human". Si no, final_output = {**parsed_output, "approved_actions": <acciones cuyo id está en approved_action_ids; o todas si no hubo HITL>}, status="completed".

3.7 De StateSnapshot a AgentExecution — orchestrator._build_execution

class AgentExecution(BaseModel):
    trace_id: UUID; agent_name: str; agent_version: str
    status: Literal["running","awaiting_approval","blocked_by_guardrail","completed","failed"]
    started_at: datetime; finished_at: datetime | None
    decision_path: list[DecisionStep]; violations: list[GuardrailViolation]
    proposed_actions: list[ProposedAction]; needs_human_for: list[ProposedAction] | None
    final_output: dict[str,Any] | None; error: str | None
class DecisionStep(BaseModel):    step: str; timestamp: datetime; duration_ms: int; detail: dict[str,Any]
class ProposedAction(BaseModel):  id: str; action: str; target: str; risk_score: int (1..5); rollback_plan: str; requires_approval: bool
class AgentExecutionSummary(BaseModel):  trace_id; agent_name; agent_version; status; started_at; finished_at; n_violations: int; n_proposed_actions: int

Reglas para derivar status a partir del StateSnapshot de LangGraph:

  1. status = state.values.get("status", "running").
  2. Si crashed y status no es terminal ⇒ status = "failed" (error = error_del_state or "internal_error").
  3. Si state.next (hay un nodo pendiente) y status no es terminal ⇒ status = "awaiting_approval" (es el interrupt() de approve_gate).
  4. needs_human_for: solo si status == "awaiting_approval" ⇒ [a for a in proposed_actions if a.risk_score >= agent_def.risk_threshold_for_hitl or a.requires_approval]; en otro caso None.
  5. started_at = decision_path[0].timestamp (o now si vacío); finished_at = now solo si status es terminal.
  6. Terminales = {"completed","failed","blocked_by_guardrail"}.

4. Cadenas de llamada por endpoint (api/)

Todos pasan primero por TraceIdMiddleware (lee/crea X-Trace-Id, lo bind-ea a structlog, lo devuelve en la respuesta). Montaje (en main.create_app): agents.router→/agents, executions.invoke_router→/agents, executions.router→/executions, policies.router→/policies, violations.router→/violations, más GET /health.

Endpoint Handler (api/…) Deps inyectadas Llama a → Escribe
POST /agents/{name}/invoke executions.invoke_agent Registry, PolicyStore, Orchestrator, Settings registry.get_agent(name) → policies.get_policy(agent.guardrails[0]) → orchestrator.invoke(agent_def, policy, body.input) → _record_execution(data_dir, trace_id, name, version) (escribe execution_index.json) → si status terminal: append_execution (executions.jsonl) → por cada violación: append_violation (violations.jsonl). 404 si el agente no existe; 422 si no tiene política.
GET /executions executions.list_executions Registry, PolicyStore, Orchestrator, Settings read_execution_summaries(data_dir) (lee executions.jsonl) + por cada trace_id en execution_index.json no presente ya: registry.get_version + policies.get_policy + orchestrator.snapshot → AgentExecutionSummary; fusiona. (Así aparecen también las awaiting_approval, que no están en el JSONL.)
GET /executions/{trace_id} executions.get_execution Registry, PolicyStore, Orchestrator, Settings _resolve(...) (lee execution_index.json → reconstruye agent_def/policy; 404 si no está, 500 si la config referida no existe) → orchestrator.snapshot(...) → 404 si None.
POST /executions/{trace_id}/approve executions.approve_execution Registry, PolicyStore, Orchestrator, Settings _resolve → _ensure_awaiting (orchestrator.snapshot; 409 si no está en awaiting_approval) → orchestrator.resume(decision={approved_action_ids: body.approved_action_ids, comment: body.comment, rejected: False}) → append_execution (executions.jsonl).
POST /executions/{trace_id}/reject executions.reject_execution Registry, PolicyStore, Orchestrator, Settings _resolve → _ensure_awaiting → orchestrator.resume(decision={approved_action_ids: [], rejected: True, reason: body.reason}) → append_execution.
GET /agents agents.list_agents Registry registry.list_agents()
GET /agents/{name} agents.get_agent Registry registry.get_agent(name) (404 si no existe)
GET /agents/{name}/versions agents.list_versions Registry registry.list_versions(name) (404)
GET /agents/{name}/versions/{version} agents.get_version Registry registry.get_version(name, version) (404)
GET /agents/{name}/versions/{v_from}/diff/{v_to} agents.diff_versions Registry registry.diff_versions(...) → DiffResult
GET /policies policies.list_policies PolicyStore store.list_policies()
GET /policies/{name}/versions policies.list_versions PolicyStore store.list_versions(name) (404)
GET /violations?trace_id=&severity= violations.list_violations Settings read_violations(data_dir) (lee violations.jsonl) + filtros opcionales por trace_id / severity.
GET /health (en main.py) — {"status":"ok"}

Modelos de petición (en executions.py): InvokeRequest{input: str, version: str | None}, ApproveRequest{approved_action_ids: list[str]=[], comment: str | None}, RejectRequest{reason: str}. Modelos de respuesta = los de domain/ (response_model=…).

Helpers internos de executions.py (índice de ejecuciones, en data_dir/execution_index.json): _index_path, _load_index, _record_execution(data_dir, trace_id, agent_name, version), _resolve(registry, policies, data_dir, trace_id) → (AgentDefinition, PolicyDefinition) (con HTTPException 404/500), _ensure_awaiting(orchestrator, agent_def, policy, trace_id) (HTTPException 404/409).

5. Persistencia: quién escribe / lee qué

Artefacto Formato Lo escribe Lo lee
agents/<n>/versions/<v>.yaml, agents/<n>/index.yaml YAML FileSystemAgentRegistry.upsert_version (a mano en este MVP) FileSystemAgentRegistry.get_* / list_* / diff_versions
policies/<n>/versions/<v>.yaml, policies/<n>/index.yaml YAML a mano FileSystemPolicyStore.get_policy / list_*
data/checkpoints.sqlite SQLite (LangGraph AsyncSqliteSaver) el grafo, en cada ainvoke/transición de nodo (incl. el interrupt()) — vía runtime/checkpointer.build_checkpointer(data_dir) orchestrator.snapshot / resume (graph.aget_state / Command(resume=…))
data/execution_index.json JSON {trace_id: {agent_name, version}} executions._record_execution — en cada invoke executions._resolve, executions.list_executions
data/executions.jsonl JSONL append-only (AgentExecution por línea) api/persistence.append_execution — en invoke si el status es terminal y siempre tras approve/reject api/persistence.read_execution_summaries (endpoint GET /executions)
data/violations.jsonl JSONL append-only (GuardrailViolation por línea) api/persistence.append_violation — en invoke, una por violación de la ejecución api/persistence.read_violations (endpoint GET /violations)

Hashing/versionado: registry/versioning.compute_hash(yaml_text) = SHA-256 del YAML con espacios finales recortados; unified_diff(a, b, label_a, label_b) = difflib.unified_diff. (En los index.yaml de ejemplo el hash es "pending" y nadie lo valida al cargar.)

6. Dashboard ↔ Core ( forja_dashboard)

El dashboard no comparte código con el core: solo lo llama por HTTP a través de CoreClient (dashboard/src/ forja_dashboard/client.py, httpx síncrono con 2 retries, base_url = FORJA_CORE_URL; mapea 404/409/422 → {"error": <json>}).

CoreClient.<método> Endpoint del core Página(s) que lo usan
health() GET /health app.py (sidebar)
list_agents() GET /agents Registro, Ejecutar
get_agent(name) GET /agents/{name} Registro
list_versions(name) GET /agents/{name}/versions Registro
diff_versions(name, a, b) GET /agents/{name}/versions/{a}/diff/{b} Registro (→ components/diff_view)
invoke_agent(name, body) POST /agents/{name}/invoke Ejecutar
list_executions() GET /executions Aprobaciones (filtra status=="awaiting_approval"), Historial
get_execution(trace_id) GET /executions/{trace_id} Aprobaciones, Historial (→ components/trace_view, violation_view)
approve(trace_id, body) POST /executions/{trace_id}/approve Aprobaciones
reject(trace_id, body) POST /executions/{trace_id}/reject Aprobaciones
list_violations(**filters) GET /violations?… Historial
list_policies() GET /policies Politicas
list_policy_versions(name) GET /policies/{name}/versions Politicas

Páginas (Streamlit multipágina; el nº y el emoji del nombre del fichero son la navegación): app.py (raíz), pages/1_🏛️_Registro.py, pages/2_▶️_Ejecutar.py, pages/3_🤝_Aprobaciones.py, pages/4_📜_Historial.py, pages/5_📐_Politicas.py. Componentes reutilizables: components/diff_view.render_unified_diff(diff_text), components/trace_view.render_trace(decision_path), components/violation_view.render_violations(violations).

7. Arranque y ciclo de vida

Proceso core (uvicorn forja_core.main:app):

  1. Import de forja_core.main ⇒ se ejecuta app = create_app(): Settings() → configure_logging(level=settings.log_level) → FastAPI(...) → app.add_middleware(TraceIdMiddleware) → registra GET /health → from forja_core.api import agents, executions, policies, violations → include_router ×5.
  2. Las dependencias (deps.get_registry, get_policy_store, get_llm_provider, get_guardrail_engine, get_orchestrator) no se construyen aún; se construyen y cachean en la primera request que las inyecta.
  3. Cada request: TraceIdMiddleware.dispatch → router → resuelve Depends(...) (que pueden disparar la construcción perezosa) → handler → respuesta con X-Trace-Id.

Contenedores (docker-compose.yml): servicio core (core/Dockerfile, uvicorn forja_core.main:app --host 0.0.0.0 --port 8000, HEALTHCHECK → curl /health, monta ./agents:ro, ./policies:ro, ./data:rw, env DATA_DIR=/app/data, AGENTS_DIR=/app/agents, POLICIES_DIR=/app/policies, env_file: .env); servicio dashboard (dashboard/Dockerfile, streamlit run app.py, HEALTHCHECK → /_stcore/health, depends_on: core: service_healthy, env FORJA_CORE_URL=http://core:8000, monta ./agents:ro para leer los examples/*.txt).

Settings (env vars) — config.py:

Campo Env var Default Lo consume
llm_provider LLM_PROVIDER mock build_llm_provider
llm_fallback_provider LLM_FALLBACK_PROVIDER "" (declarado; la factory aún no lo usa)
azure_openai_* AZURE_OPENAI_* "" / 2024-08-01-preview AzureOpenAIProvider
openai_api_key / openai_model OPENAI_API_KEY / OPENAI_MODEL "" / gpt-4o OpenAIProvider
guardrails_nemo_enabled GUARDRAILS_NEMO_ENABLED False build_guardrail_engine
log_level LOG_LEVEL INFO configure_logging
data_dir DATA_DIR ./data build_checkpointer, _record_execution, append_*, read_*
agents_dir AGENTS_DIR ./agents build_agent_registry
policies_dir POLICIES_DIR ./policies build_policy_store

8. Aristas y "gotchas"

  • llm_fallback_provider: existe en Settings y en .env.example, pero build_llm_provider aún no lo aplica (queda como punto de extensión).
  • build_llm_provider usa match sin case _:; al ser el tipo un Literal de tres valores es exhaustivo, pero un valor inesperado caería en "ninguna rama".
  • NeMo: NeMoGuardrailsEngine solo entra al CompositeGuardrailEngine si GUARDRAILS_NEMO_ENABLED=true; por defecto el composite tiene un solo engine.
  • detect_pii se comporta distinto según el entorno: con presidio-analyzer instalado (imagen Docker) usa Presidio + en_core_web_sm; sin él (venv local típico), regex fallback. La política default pide solo recognizers de patrón (EMAIL_ADDRESS, ES_NIF, IP_ADDRESS, IBAN_CODE) para evitar falsos positivos del NER.
  • AsyncSqliteSaver liga su conexión aiosqlite al event loop activo: por eso build_checkpointer es un async context manager y el orchestrator lo abre y cierra en cada operación (no se reusa entre llamadas). Requiere aiosqlite<0.21.
  • data/ debe existir y ser escribible por el proceso/usuario del contenedor (agent, uid 1000); si no, invoke fallará al escribir el log/checkpoint.
  • make test / make lint / make smoke asumen que el venv está activado (los binarios pytest/ruff/mypy/docker en el PATH).
  • tests/integration/ monta la app con TestClient apuntando a los assets reales del repo (agents/, policies/); tests/unit/test_api_* usan tests/fixtures/ (versiones mínimas). Ambos hacen deps.*.cache_clear().
Reference in New Issue View Git Blame Copy Permalink