LangGraph API¶

Complete API reference for LangGraph components.

State Classes¶

AgentState¶

class AgentState(TypedDict):
    """Workflow state container."""

    query: str                           # User query
    user_id: Optional[str]               # User identifier
    chat_id: Optional[str]               # Chat session ID
    chat_history: List[Dict[str, str]]   # Previous messages
    validation_result: Optional[dict]    # Validation outcome
    search_results: List[dict]           # Search results
    tool_decisions: List[dict]           # Tool decisions
    tool_results: List[dict]             # Tool execution results
    final_response: str                  # Generated response
    metadata: Dict[str, Any]             # Additional metadata
    status: str                          # Workflow status

Metadata Fields:

The metadata dict supports the following optional keys:

Key	Type	Description
`_skip_token_tracking`	`bool`	Skip token usage tracking (default: `False`)
`agent_id`	`str`	Agent identifier for token tracking
`model_id`	`str`	Model identifier

WorkflowConfig¶

@dataclass
class WorkflowConfig:
    """Configuration for LangGraph workflows."""

    enable_validation: bool = True     # Enable validation node
    enable_tools: bool = False         # Enable tool nodes
    enable_tracing: bool = False       # Enable LangSmith tracing
    max_iterations: int = 10           # Max workflow iterations
    temperature: float = 0.1           # LLM temperature

WorkflowStatus¶

class WorkflowStatus(Enum):
    """Workflow execution status."""

    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"

NodeResult¶

@dataclass
class NodeResult:
    """Result from a workflow node."""

    success: bool                      # Execution success
    data: Dict[str, Any]               # Result data
    next_node: Optional[str] = None    # Next node to execute
    error: Optional[str] = None        # Error message if failed

Workflow Classes¶

BaseAgentWorkflow¶

class BaseAgentWorkflow:
    """Base class for LangGraph workflows."""

Constructor¶

BaseAgentWorkflow(
    name: str,
    config: Optional[WorkflowConfig] = None,
    llm_service: Optional[LLMServiceInterface] = None,
)

Methods¶

execute¶

def execute(self, initial_state: Dict[str, Any]) -> Dict[str, Any]

Execute the workflow synchronously.

execute_async¶

async def execute_async(self, initial_state: Dict[str, Any]) -> Dict[str, Any]

Execute the workflow asynchronously.

stream¶

async def stream(self, initial_state: Dict[str, Any]) -> AsyncIterator[str]

Stream workflow execution results.

ToolUseWorkflow¶

class ToolUseWorkflow(BaseAgentWorkflow):
    """Workflow for intelligent tool selection and execution."""

Constructor¶

ToolUseWorkflow(
    name: str,
    tools: List[Tool] = None,
    config: Optional[WorkflowConfig] = None,
    validator: Optional[ValidatorInterface] = None,
    llm_service: Optional[LLMServiceInterface] = None,
    search_handler: Optional[SearchHandlerInterface] = None,
)

Parameter	Type	Default	Description
`name`	`str`	-	Workflow name
`tools`	`List[Tool]`	`None`	Available tools
`config`	`WorkflowConfig`	`None`	Configuration
`validator`	`ValidatorInterface`	`None`	Input validator
`llm_service`	`LLMServiceInterface`	`None`	LLM service
`search_handler`	`SearchHandlerInterface`	`None`	Search handler

Workflow Graph¶

graph TD
    A[Start] --> B[Validation]
    B -->|Valid| C[Tool Decision]
    B -->|Invalid| G[End]
    C -->|Use Tools| D[Tool Execution]
    C -->|No Tools| E[Response Generation]
    D --> E
    E --> F[End]

MultiAgentWorkflow¶

class MultiAgentWorkflow(BaseAgentWorkflow):
    """Workflow for coordinating multiple agents."""

Constructor¶

MultiAgentWorkflow(
    name: str,
    agents: List[AgentInterface],
    config: Optional[WorkflowConfig] = None,
)

Node Classes¶

BaseNode¶

class BaseNode:
    """Base class for workflow nodes."""

Constructor¶

BaseNode(name: str)

Methods¶

process¶

def process(self, state: AgentState) -> NodeResult

Process the state and return result. Override in subclasses.

ValidationNode¶

class ValidationNode(BaseNode):
    """Node for input validation."""

Constructor¶

ValidationNode(validator: ValidatorInterface)

SearchNode¶

class SearchNode(BaseNode):
    """Node for knowledge base search."""

Constructor¶

SearchNode(search_handler: SearchHandlerInterface)

ToolDecisionNode¶

class ToolDecisionNode(BaseNode):
    """Node for deciding which tools to use."""

Constructor¶

ToolDecisionNode(
    tools: List[Tool],
    llm_service: LLMServiceInterface,
)

ToolExecutionNode¶

class ToolExecutionNode(BaseNode):
    """Node for executing selected tools."""

Constructor¶

ToolExecutionNode(tools: List[Tool])

ResponseGenerationNode¶

class ResponseGenerationNode(BaseNode):
    """Node for generating LLM responses."""

Constructor¶

ResponseGenerationNode(llm_service: LLMServiceInterface)

StreamingResponseNode¶

class StreamingResponseNode(BaseNode):
    """Node for streaming response generation."""

CoordinatorNode¶

class CoordinatorNode(BaseNode):
    """Node for coordinating multi-agent tasks."""

AggregationNode¶

class AggregationNode(BaseNode):
    """Node for aggregating multi-agent results."""

Routing Classes¶

RoutingCondition¶

@dataclass
class RoutingCondition:
    """Condition for routing decisions."""

    name: str                              # Condition name
    check: Callable[[dict], bool]          # Check function

Example¶

needs_tools = RoutingCondition(
    name="needs_tools",
    check=lambda state: bool(state.get("tool_decisions"))
)

RoutingRule¶

@dataclass
class RoutingRule:
    """Rule for routing to a target node."""

    condition: RoutingCondition    # Condition to check
    target: str                    # Target node name
    priority: int = 0              # Rule priority

ConditionalRouter¶

class ConditionalRouter:
    """Router for conditional workflow paths."""

Constructor¶

ConditionalRouter(
    rules: List[RoutingRule],
    default_target: str = "end",
)

Methods¶

route¶

def route(self, state: dict) -> str

Determine the next node based on state.

RoutingStrategy Classes¶

class RoutingStrategy:
    """Base class for routing strategies."""

class PriorityBasedRoutingStrategy(RoutingStrategy):
    """Route based on rule priority."""

class WeightedRoutingStrategy(RoutingStrategy):
    """Route with weighted random selection."""

class AdaptiveRoutingStrategy(RoutingStrategy):
    """Adapt routing based on past performance."""

Orchestration Classes¶

AgentCapability¶

@dataclass
class AgentCapability:
    """Defines an agent's capabilities."""

    name: str                      # Capability name
    description: str               # Description
    domains: List[str] = []        # Expertise domains

AgentProfile¶

@dataclass
class AgentProfile:
    """Profile for an agent in orchestration."""

    agent_id: str
    agent: AgentInterface
    capabilities: List[AgentCapability]
    role: AgentRole = AgentRole.WORKER

AgentRegistry¶

class AgentRegistry:
    """Registry for managing agents."""

Methods¶

register_agent¶

def register_agent(
    self,
    agent_id: str,
    agent: AgentInterface,
    capabilities: List[AgentCapability] = None,
) -> None

get_agent¶

def get_agent(self, agent_id: str) -> Optional[AgentProfile]

find_agents_for_capability¶

def find_agents_for_capability(self, capability_name: str) -> List[AgentProfile]

Orchestrator Classes¶

class BaseOrchestrator:
    """Base orchestrator class."""

class CoordinatorOrchestrator(BaseOrchestrator):
    """Coordinator pattern orchestrator."""

class PeerToPeerOrchestrator(BaseOrchestrator):
    """Peer-to-peer pattern orchestrator."""

class HierarchicalOrchestrator(BaseOrchestrator):
    """Hierarchical pattern orchestrator."""

OrchestratedMultiAgentWorkflow¶

class OrchestratedMultiAgentWorkflow:
    """Workflow with advanced multi-agent orchestration."""

Constructor¶

OrchestratedMultiAgentWorkflow(orchestrator: BaseOrchestrator)

Methods¶

execute¶

async def execute(
    self,
    query: str,
    context: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]

A2A Protocol Classes¶

MessageType¶

class MessageType(Enum):
    """Types of A2A messages."""

    REQUEST = "request"
    RESPONSE = "response"
    BROADCAST = "broadcast"
    STATUS = "status"

MessagePriority¶

class MessagePriority(Enum):
    """Message priority levels."""

    LOW = "low"
    NORMAL = "normal"
    HIGH = "high"
    CRITICAL = "critical"

A2AMessage¶

@dataclass
class A2AMessage:
    """Agent-to-Agent protocol message."""

    message_type: MessageType
    sender_id: str
    receiver_id: str
    content: Any
    metadata: A2AMetadata

A2AProtocolLayer¶

class A2AProtocolLayer:
    """Protocol layer for A2A communication."""

Methods¶

send¶

async def send(self, message: A2AMessage) -> A2AMessage

broadcast¶

async def broadcast(
    self,
    sender_id: str,
    content: Any,
    receivers: List[str],
) -> List[A2AMessage]

Factory Functions¶

create_coordinator_orchestrator¶

def create_coordinator_orchestrator(
    coordinator_id: str,
    registry: AgentRegistry,
) -> CoordinatorOrchestrator

create_peer_to_peer_orchestrator¶

def create_peer_to_peer_orchestrator(
    registry: AgentRegistry,
) -> PeerToPeerOrchestrator

create_hierarchical_orchestrator¶

def create_hierarchical_orchestrator(
    registry: AgentRegistry,
) -> HierarchicalOrchestrator

create_priority_router¶

def create_priority_router(
    rules: List[RoutingRule],
) -> ConditionalRouter

create_weighted_router¶

def create_weighted_router(
    rules: List[RoutingRule],
    weights: Dict[str, float],
) -> ConditionalRouter

create_adaptive_router¶

def create_adaptive_router(
    rules: List[RoutingRule],
) -> ConditionalRouter