Tactical blueprint for stateful AI agents and multi-agent systems. Focuses on LangGraph orchestration, state persistence, and procedural loop logic.
This blueprint provides the procedural truth for engineering resilient, stateful AI agents using LangGraph and LangChain 1.x.
This skill should be used when completing tasks related to agentic loop mastery.
Follow these procedures to implement the capability:
TypedDict or Pydantic.Annotated with reducers (e.g., add_messages) for fields that accumulate data over time.interaction_log list in the state to record reasoning steps that are not part of the message history.agent-staffing.json mission.conditional_edges to route the state to specialists. Specialists update the state and route back to the Supervisor.interrupt_before=["tool_execution_node"] for high-risk tools (e.g., file deletion, payments).is_approved boolean to the state.| Symptom | Probable Cause | Recovery Operation |
| :--- | :--- | :--- |
| State Drift | Node functions modifying state outside of the return statement. | Run an "Immutability Audit"; ensure all nodes are pure functions that return state updates; use copy.deepcopy if necessary. |
| Loop Stalling | Agent repeating the same tool call with same result. | Trigger the "Anti-Loop" gate: detect duplicate tool_calls in history; force a "Strategy Shift" node that requires a new plan. |
| Persistence Sync Error | SQLite/Postgres connection failure. | Implement a "Persistence Fallback" to in-memory Checkpointer; notify the system orchestrator for a "Health Restart." |
def supervisor_node(state: AgentState):
prompt = f"Objective: {state['objective']}. Specialists available: {get_staffing_registry()}."
# Use .with_structured_output to force a valid 'next_agent' decision
decision = llm.with_structured_output(SupervisionSchema).invoke(prompt)
return {"messages": [SystemMessage(content=f"Delegating to {decision.next_agent}")], "current_specialist": decision.next_agent}
def final_verification(state: AgentState):
if state["iteration_count"] > 10:
return {"messages": [AIMessage(content="CRITICAL: Goal unreached within safety limit. Terminating chain.")]}
# ... regular verification logic
| Action | Tool / Command |
| :--- | :--- |
| Visualize SOP | app.get_graph().print_ascii() |
| Time Travel | app.get_state_history(thread_id) |
| Monitor State | LangSmith (inspect 'checkpoint' metadata) |
Before finalizing an agentic loop:
max_iterations counter is implemented in the state.Category:other