from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
current_step: str
iteration: int
def agent_node(state: AgentState) -> AgentState:
return {"current_step": "processed", "iteration": state["iteration"] + 1}
def tool_node(state: AgentState) -> AgentState:
return {"current_step": "tools_executed"}
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_edge("agent", "tools")
graph.add_conditional_edges(
"tools",
lambda state: "end" if state["iteration"] >= 3 else "continue",
{"end": END, "continue": "agent"}
)
app = graph.compile()
def router(state: AgentState) -> str:
"""Route based on state conditions."""
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
elif state["iteration"] >= state.get("max_iterations", 10):
return "end"
else:
return "agent"
graph.add_conditional_edges(
"agent",
router,
{
"tools": "tool_executor",
"agent": "agent",
"end": END
}
)
from langgraph.checkpoint.sqlite import SqliteSaver
memory = SqliteSaver.from_conn_string(":memory:")
app = graph.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "conversation-1"}}
result = app.invoke(initial_state, config=config)
result = app.invoke(None, config=config)
from langgraph.graph import StateGraph
graph = StateGraph(AgentState)
app = graph.compile(
checkpointer=memory,
interrupt_before=["tool_executor"]
)
result = app.invoke(initial_state, config)
result = app.invoke(None, config)
const langgraphStateGraphTask = defineTask({
name: 'langgraph-state-graph-design',
description: 'Design and implement a LangGraph StateGraph workflow',
inputs: {
workflowName: { type: 'string', required: true },
stateSchema: { type: 'object', required: true },
nodes: { type: 'array', required: true },
edges: { type: 'array', required: true },
enablePersistence: { type: 'boolean', default: true },
interruptPoints: { type: 'array', default: [] }
},
outputs: {
graphCode: { type: 'string' },
stateSchemaCode: { type: 'string' },
compiledGraph: { type: 'boolean' },
artifacts: { type: 'array' }
},
async run(inputs, taskCtx) {
return {
kind: 'skill',
title: `Design StateGraph: ${inputs.workflowName}`,
skill: {
name: 'langgraph-state-graph',
context: {
workflowName: inputs.workflowName,
stateSchema: inputs.stateSchema,
nodes: inputs.nodes,
edges: inputs.edges,
enablePersistence: inputs.enablePersistence,
interruptPoints: inputs.interruptPoints,
instructions: [
'Analyze workflow requirements and state needs',
'Design state schema with proper typing',
'Create node functions with state transformations',
'Define edges and conditional routing logic',
'Configure persistence if enabled',
'Add interrupt points for human-in-the-loop',
'Compile and validate the graph'
]
}
},
io: {
inputJsonPath: `tasks/${taskCtx.effectId}/input.json`,
outputJsonPath: `tasks/${taskCtx.effectId}/result.json`
}
};
}
});
