Copilot instructions

1---
2applyTo: "**/langgraph/**, **/graphs/**, **/workflows/**/*.py"
3description: "LangGraph StateGraph construction, node functions, edge types, checkpointing, HITL patterns, streaming, subgraphs, and Bedrock integration."
4---
5 
6# LangGraph — Copilot Instructions
7 
8> Applied automatically when working with LangGraph graph files, state definitions, and Python workflow modules. Loaded alongside copilot-instructions.md.
9 
10---
11 
12## StateGraph Construction
13 
14Always define state as a `TypedDict` with field-level `Annotated` reducers. Use `operator.add` for list fields that accumulate across nodes; use default assignment (`field: str`) for scalar fields that overwrite.
15 
16```python
17from typing import TypedDict, Annotated, Sequence
18import operator
19from langchain_core.messages import BaseMessage
20 
21class AgentState(TypedDict):
22    messages: Annotated[Sequence[BaseMessage], operator.add]  # Accumulates
23    user_id: str                                               # Overwrites
24    current_step: str                                          # Overwrites
25    error: str | None                                          # Overwrites; None = no error
26    context: Annotated[list[str], operator.add]               # Accumulates retrieved context
27```
28 
29### Graph Construction Pattern
30 
31```python
32from langgraph.graph import StateGraph, END
33 
34# Build graph
35builder = StateGraph(AgentState)
36 
37# Add nodes
38builder.add_node("retrieve", retrieve_node)
39builder.add_node("generate", generate_node)
40builder.add_node("validate", validate_node)
41builder.add_node("human_review", human_review_node)
42 
43# Set entry point
44builder.set_entry_point("retrieve")
45 
46# Add fixed edges
47builder.add_edge("retrieve", "generate")
48builder.add_edge("human_review", "generate")
49 
50# Add conditional edges
51builder.add_conditional_edges(
52    "generate",
53    route_after_generate,           # routing function — returns a string key
54    {
55        "valid": "validate",
56        "needs_review": "human_review",
57        "error": END,
58    }
59)
60 
61builder.add_edge("validate", END)
62 
63# Compile with checkpointer
64graph = builder.compile(checkpointer=checkpointer)
65```
66 
67---
68 
69## Node Function Requirements
70 
71Every node function must:
72- Accept the full state dict as input
73- Return a **dict** with only the keys it updates (LangGraph merges partial updates)
74- Never mutate the input state directly
75- Handle exceptions internally and set the `error` field rather than raising
76 
77```python
78from typing import Any
79 
80def generate_node(state: AgentState) -> dict[str, Any]:
81    """Generate a response based on retrieved context."""
82    try:
83        response = llm.invoke([
84            SystemMessage(content="You are a helpful assistant."),
85            HumanMessage(content=state["messages"][-1].content),
86        ])
87        return {
88            "messages": [response],       # Appended (Annotated list)
89            "current_step": "generated",  # Overwrites
90            "error": None,                # Clear any previous error
91        }
92    except Exception as exc:
93        return {
94            "error": str(exc),
95            "current_step": "error",
96        }
97```
98 
99---
100 
101## Edge Types
102 
103| Edge Type | Method | Usage |
104|-----------|--------|-------|
105| Fixed edge | `add_edge(source, target)` | Always transition from source to target |
106| Conditional edge | `add_conditional_edges(source, fn, mapping)` | Route based on state; routing function returns a string key |
107| Entry point | `set_entry_point(node)` | First node to execute |
108| Conditional entry | `set_conditional_entry_point(fn, mapping)` | Choose starting node based on initial state |
109 
110### Routing Function Pattern
111 
112```python
113def route_after_generate(state: AgentState) -> str:
114    """Route based on confidence and error state."""
115    if state.get("error"):
116        return "error"
117    last_message = state["messages"][-1]
118    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
119        return "tools"
120    if state.get("requires_human_approval"):
121        return "needs_review"
122    return "valid"
123```
124 
125---
126 
127## MessagesState for Conversation Graphs
128 
129Use `MessagesState` from LangGraph for chat applications — it pre-defines the `messages` field with the correct `add_messages` reducer that handles deduplication by message ID.
130 
131```python
132from langgraph.graph import MessagesState
133 
134class ChatState(MessagesState):
135    # MessagesState provides: messages: Annotated[list[AnyMessage], add_messages]
136    session_id: str
137    user_context: dict[str, str]
138 
139# The add_messages reducer automatically handles:
140# - Deduplication (updates existing message if same ID)
141# - Appending new messages
142# - Proper serialization for checkpointing
143```
144 
145---
146 
147## Checkpointer Setup
148 
149### SQLite (development and single-instance production)
150 
151```python
152from langgraph.checkpoint.sqlite import SqliteSaver
153from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
154 
155# Synchronous
156with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
157    graph = builder.compile(checkpointer=checkpointer)
158 
159# Asynchronous
160async with AsyncSqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
161    graph = builder.compile(checkpointer=checkpointer)
162```
163 
164### Redis (production, multi-instance)
165 
166```python
167from langgraph.checkpoint.redis import RedisSaver
168from redis import Redis
169 
170redis_client = Redis.from_url(
171    url=os.environ["REDIS_URL"],          # redis://redis-host:6379/0
172    decode_responses=False,                # Must be False for LangGraph
173    socket_timeout=5,
174    socket_connect_timeout=5,
175)
176 
177checkpointer = RedisSaver(redis_client)
178checkpointer.setup()  # Creates required key structures — call once on startup
179graph = builder.compile(checkpointer=checkpointer)
180```
181 
182### Thread Config — Required for Checkpointing
183 
184```python
185# Every invocation that uses checkpointing MUST provide a thread_id
186config = {"configurable": {"thread_id": "user-session-abc123"}}
187 
188result = graph.invoke({"messages": [HumanMessage(content="Hello")]}, config=config)
189 
190# Resume from checkpoint (HITL pattern)
191graph.invoke(None, config=config)  # None input = resume from last checkpoint
192```
193 
194---
195 
196## Human-in-the-Loop (HITL)
197 
198### Interrupt Before/After a Node
199 
200```python
201# interrupt_before: pause BEFORE executing the named node
202graph = builder.compile(
203    checkpointer=checkpointer,
204    interrupt_before=["human_review"],   # Graph pauses before this node runs
205    interrupt_after=["generate"],         # Graph pauses after this node completes
206)
207 
208# Execute until interrupt
209result = graph.invoke(
210    {"messages": [HumanMessage(content=user_input)]},
211    config={"configurable": {"thread_id": thread_id}}
212)
213 
214# Inspect state at interrupt point
215snapshot = graph.get_state(config={"configurable": {"thread_id": thread_id}})
216pending_messages = snapshot.values["messages"]
217 
218# Human updates state and resumes
219graph.update_state(
220    config={"configurable": {"thread_id": thread_id}},
221    values={"human_approved": True, "reviewer_id": "emp_456"},
222    as_node="human_review",
223)
224 
225# Resume execution — pass None as input to continue from checkpoint
226final_result = graph.invoke(None, config={"configurable": {"thread_id": thread_id}})
227```
228 
229---
230 
231## Streaming
232 
233### `.stream()` — Node-level events
234 
235```python
236for chunk in graph.stream(
237    {"messages": [HumanMessage(content="Analyse this document")]},
238    config={"configurable": {"thread_id": thread_id}},
239    stream_mode="values",   # "values" = full state after each node; "updates" = partial update dict
240):
241    print(chunk)
242```
243 
244### `.astream_events()` — Token-level streaming
245 
246```python
247async for event in graph.astream_events(
248    {"messages": [HumanMessage(content="Summarise")]},
249    config={"configurable": {"thread_id": thread_id}},
250    version="v2",
251):
252    if event["event"] == "on_chat_model_stream":
253        chunk = event["data"]["chunk"]
254        print(chunk.content, end="", flush=True)
255    elif event["event"] == "on_chain_end":
256        print(f"\n[Node completed: {event['name']}]")
257```
258 
259---
260 
261## Subgraph Pattern
262 
263Parent and child graphs must have compatible state schemas. The child graph can only update keys present in the parent state.
264 
265```python
266# Child graph state — must be a subset of parent state keys
267class SubState(TypedDict):
268    messages: Annotated[Sequence[BaseMessage], operator.add]
269    sub_result: str
270 
271child_builder = StateGraph(SubState)
272child_builder.add_node("sub_process", sub_process_node)
273child_builder.set_entry_point("sub_process")
274child_builder.add_edge("sub_process", END)
275child_graph = child_builder.compile()
276 
277# Add compiled child graph as a node in parent graph
278parent_builder.add_node("run_subgraph", child_graph)
279parent_builder.add_edge("preprocess", "run_subgraph")
280parent_builder.add_edge("run_subgraph", "postprocess")
281```
282 
283---
284 
285## LangGraph Studio Integration
286 
287`langgraph.json` at repo root:
288 
289```json
290{
291  "dependencies": ["."],
292  "graphs": {
293    "main_graph": "./src/graphs/main.py:graph",
294    "research_graph": "./src/graphs/research.py:graph"
295  },
296  "env": ".env",
297  "python_version": "3.12"
298}
299```
300 
301---
302 
303## LangGraph + AWS Bedrock Claude Integration
304 
305```python
306from langchain_aws import ChatBedrock
307import boto3
308 
309bedrock_client = boto3.client(
310    "bedrock-runtime",
311    region_name="eu-west-1",
312)
313 
314llm = ChatBedrock(
315    model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
316    client=bedrock_client,
317    model_kwargs={
318        "max_tokens": 4096,
319        "temperature": 0.0,    # 0.0 for deterministic tool-calling graphs
320        "top_p": 0.999,
321    },
322    streaming=True,             # Enable for token-level streaming in astream_events
323)
324 
325# Bind tools to the model
326llm_with_tools = llm.bind_tools(tools=[search_tool, calculator_tool])
327 
328def agent_node(state: AgentState) -> dict:
329    response = llm_with_tools.invoke(state["messages"])
330    return {"messages": [response]}
331```
332 
333---
334 
335## Common Graph Topologies
336 
337| Topology | Pattern | Use Case |
338|----------|---------|---------|
339| Linear | A → B → C → END | Simple sequential processing |
340| Branching | A → conditional → {B, C} → END | Route based on content/confidence |
341| Loop | A → B → conditional → {A (retry), END} | Retry until success or max iterations |
342| Supervisor | Supervisor → conditional → {Worker1, Worker2} → Supervisor | Multi-agent orchestration |
343| RAG | Retrieve → Grade → conditional → {Generate, Rewrite} | Self-correcting RAG |
344 
345### Loop with Max Iterations Guard
346 
347```python
348class LoopState(TypedDict):
349    messages: Annotated[Sequence[BaseMessage], operator.add]
350    iteration_count: int
351 
352def increment_counter(state: LoopState) -> dict:
353    return {"iteration_count": state["iteration_count"] + 1}
354 
355def should_continue(state: LoopState) -> str:
356    if state["iteration_count"] >= 5:
357        return "max_iterations_reached"
358    if is_complete(state):
359        return "done"
360    return "continue"
361```
362 
363---
364 
365## Error Handling in Nodes
366 
367```python
368def risky_node(state: AgentState) -> dict[str, Any]:
369    try:
370        result = external_api_call(state["context"])
371        return {"result": result, "error": None}
372    except httpx.TimeoutException:
373        return {"error": "API timeout after 30s", "current_step": "retry_needed"}
374    except httpx.HTTPStatusError as exc:
375        if exc.response.status_code == 429:
376            return {"error": "rate_limited", "current_step": "backoff_needed"}
377        return {"error": f"HTTP {exc.response.status_code}", "current_step": "error"}
378```
379 
380---
381 
382## Memory Management
383 
384| Memory Type | Implementation | Scope |
385|------------|---------------|-------|
386| Short-term (in-conversation) | LangGraph state dict | Single thread/session |
387| Cross-session (user profile) | `store` parameter in `compile(store=InMemoryStore())` | Shared across threads by user_id |
388| Long-term (document/knowledge) | Tool — vector store search via LangChain retriever | Accessed by any graph node |
389 
390```python
391from langgraph.store.memory import InMemoryStore
392# In production: use RedisStore or PostgresStore
393 
394store = InMemoryStore()
395graph = builder.compile(checkpointer=checkpointer, store=store)
396 
397# In a node — access cross-session memory
398def personalized_node(state: AgentState, store: BaseStore) -> dict:
399    user_prefs = store.get(("user_preferences",), state["user_id"])
400    # ...
401```
402