agentic_fabric.runners.langgraph_runner

LangGraph runner implementation.

LangGraph excels at:

  • Complex conditional flows

  • State management

  • Cycles and loops

  • Integration with LangChain ecosystem

Module Contents

Classes

LangGraphRunner

Runner that uses LangGraph for fabric agent execution.

API

class agentic_fabric.runners.langgraph_runner.LangGraphRunner

Bases: agentic_fabric.runners.base.BaseRunner

Runner that uses LangGraph for fabric agent execution.

Initialization

Initialize LangGraph runner.

framework_name = 'langgraph'
build_fabric_agent(fabric_agent_config: dict[str, Any]) → Any

Build a LangGraph ReAct agent from fabric agent configuration.

The current adapter condenses the universal fabric agent configuration into a LangGraph prebuilt ReAct agent with the declared tools attached.

For single-agent configs, this produces a single ReAct agent. For multi-agent configs, each agent is built as a separate ReAct agent and tasks are wired into the system prompt as context. A full StateGraph with per-task nodes is a future enhancement.

Args: fabric_agent_config: Universal fabric agent configuration.

Returns: LangGraph runnable agent.

run(fabric_agent: Any, inputs: dict[str, Any]) → str

Execute the LangGraph workflow.

Args: fabric_agent: Compiled LangGraph. inputs: Inputs for the workflow.

Returns: Workflow output as string.

get_llm(model: str | None = None) → Any

Get LangChain-compatible LLM.

Supports Ollama via OLLAMA_BASE_URL env var, Anthropic via ANTHROPIC_API_KEY, and OpenRouter via OPENROUTER_API_KEY.

Args: model: Optional model name override.

Returns: LangChain ChatModel (ChatOllama, ChatAnthropic, or ChatOpenAI).

build_agent(agent_config: dict[str, Any], tools: list | None = None) → Any

Build a LangGraph-compatible agent.

Args: agent_config: Agent configuration. tools: Optional tools.

Returns: LangGraph agent.

build_task(task_config: dict[str, Any], agent: Any) → Any

Build a task representation for LangGraph.

In LangGraph, tasks are typically represented as graph nodes or prompts to agents. Returns a dict for now.

Args: task_config: Task configuration. agent: Agent for the task.

Returns: Task configuration dict with agent reference.

build_and_run(fabric_agent_config: dict[str, Any], inputs: dict[str, Any] | None = None) → str
agent_capabilities: ClassVar[collections.abc.Mapping[str, agentic_fabric.capabilities.AgentCapabilitySpec]] = None
agent_capability_methods: ClassVar[collections.abc.Mapping[str, str]] = None
classmethod list_capabilities(*, kind: str | None = None) → tuple[agentic_fabric.capabilities.AgentCapabilitySpec, ...]
call_capability(name: str, *args: Any, **kwargs: Any) → Any
capability_map() → collections.abc.Mapping[str, agentic_fabric.capabilities.AgentCapabilitySpec]