cleveragents-core/.opencode/skills/cleveragents-contributing/references/langchain-langgraph

LangChain / LangGraph — CleverAgents Project

⚠️ Rules here override cleverthis-guidelines. Apply these exactly.

---

Graph Design Patterns

State Definition

Always use TypedDict for state — never plain dicts or Any-typed structures:

from typing import TypedDict

class PlanGenerationState(TypedDict):
    """State for the plan generation workflow."""
    description: str           # user-provided requirements
    analysis: str              # output from analyze_requirements node
    plan: str                  # output from generate_plan node
    error: str                 # populated if any node fails
    is_complete: bool          # workflow completion flag

Rules:

• Every field must be typed — no Any
• Include an error: str field in every state for graceful failure tracking
• Document each field with a comment

Node Naming

Nodes use descriptive verb-based names — always a verb phrase:

# CORRECT:
analyze_requirements
generate_plan
validate_output
apply_changes
summarize_results

# WRONG:
node1
handler
process
step_a

Checkpointing

ALWAYS integrate MemorySaver for workflow resumption:

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

Never compile a graph without a checkpointer if the graph has more than one step.

Conditional Edges

Use clear, descriptively named decision functions:

def should_retry_or_fail(state: PlanGenerationState) -> str:
    """Decide whether to retry analysis or report failure."""
    if state.get("error") and state.get("retry_count", 0) < 3:
        return "retry"
    elif state.get("error"):
        return "fail"
    return "continue"

---

LangChain Integration

Provider Abstraction (mandatory)

# CORRECT — uses BaseLanguageModel abstraction:
from langchain_core.language_models import BaseLanguageModel

class PlanService:
    def __init__(self, llm: BaseLanguageModel) -> None:
        self.llm = llm

# WRONG — hardcodes a specific provider:
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4")   # ← never hardcode in production

Always accept BaseLanguageModel or BaseLLM as a dependency — never instantiate a specific provider in production code.

Prompt Templates

from langchain_core.prompts import ChatPromptTemplate

# CORRECT:
prompt = ChatPromptTemplate.from_template(
    "Analyze these requirements: {requirements}"
)

# WRONG — f-strings are not composable or traceable:
prompt = f"Analyze these requirements: {requirements}"

Always use ChatPromptTemplate or PromptTemplate. Never construct prompts with f-strings or string concatenation.

Sync + Async (both required)

Every chain or node must implement both:

# Sync:
result = chain.invoke({"requirements": state["description"]})

# Async (preferred for production graph nodes):
result = await chain.ainvoke({"requirements": state["description"]})

Memory

from langchain.memory import ConversationBufferMemory, EntityMemory

# Use ConversationBufferMemory for conversation history
# Use EntityMemory for entity tracking across turns

Output Parsing

from langchain_core.output_parsers import StrOutputParser, JsonOutputParser

# For plain text output:
chain = prompt | llm | StrOutputParser()

# For structured JSON output:
chain = prompt | llm | JsonOutputParser()

---

RxPY — Reactive Streams for Event Routing

RxPY is used for real-time event routing between actors and graphs, NOT inside a single LangGraph node.

Subject types — choose the right one

Subject type	When to use	Example
Subject	Hot stream — late subscribers miss earlier events	Plan phase change events
BehaviorSubject	Hot stream + late subscribers get the last emitted value	Current automation profile
ReplaySubject(n)	Buffers the last N events for late subscribers	Session history replay (last 10 events)

from rx.subject import Subject, BehaviorSubject, ReplaySubject

# Plan events (late subscriber doesn't need history)
plan_events: Subject = Subject()

# Current automation profile (late subscriber needs to know current state)
current_profile: BehaviorSubject = BehaviorSubject("manual")

# Session replay (late subscriber gets last 50 events for context)
session_replay: ReplaySubject = ReplaySubject(buffer_size=50)

Key operators

Operator	Use case
map(transform)	Transform events before routing
filter(predicate)	Route only events matching a condition
flat_map(fn)	Chain async operations returning observables
debounce(n)	Ignore events within N ms of each other (burst suppression)
throttle(n)	Allow at most one event per N ms (rate limiting)
scan(accumulator, seed)	Accumulate state across events (running totals, history)

from rx import operators as ops

plan_events.pipe(
    ops.filter(lambda e: e.phase == "execute"),
    ops.map(lambda e: e.plan_id),
    ops.throttle_first(1.0),           # at most one per second
).subscribe(lambda plan_id: trigger_checkpoint(plan_id))

Backpressure management

# PROBLEM: actor emits tokens faster than TUI can render
actor_output.pipe(
    ops.debounce(0.05),    # collapse bursts within 50ms into one event
).subscribe(update_tui)

# PROBLEM: high-frequency file system events
file_watcher.pipe(
    ops.throttle_first(1.0),   # at most one re-index trigger per second
).subscribe(trigger_reindex)

What RxPY is NOT for

• Within a single LangGraph node: use async/await and LangGraph's built-in streaming
• Database writes: use SQLAlchemy's unit-of-work pattern
• HTTP requests: use httpx directly or LangChain's built-in retry decorators

---

Configuration Rules

Environment Variables

# Standard LangChain env vars — use these, don't invent custom ones:
LANGCHAIN_TRACING_V2=true       # enable LangSmith tracing
LANGCHAIN_API_KEY=...           # LangSmith API key
LANGCHAIN_PROJECT=...           # LangSmith project name

Observability

LangSmith/OpenTelemetry must be DISABLED by default. Users opt in via environment variables — never enable automatically in production code.

# Correct: read from environment, default off:
import os
tracing_enabled = os.getenv("LANGCHAIN_TRACING_V2", "false").lower() == "true"

Provider Configuration

Support multiple providers via configuration:

# In configuration:
llm_provider: str = "openai"   # or "anthropic", "ollama", etc.
llm_model: str = "gpt-4o"

# In code — resolved at startup, never hardcoded:
def create_llm(config: LLMConfig) -> BaseLanguageModel:
    if config.llm_provider == "openai":
        return ChatOpenAI(model=config.llm_model)
    elif config.llm_provider == "anthropic":
        return ChatAnthropic(model=config.llm_model)
    raise ValueError(f"Unknown provider: {config.llm_provider}")

Retry Logic

Use LangChain's built-in retry decorators — do not implement manual retry loops:

from langchain_core.runnables import RunnableRetry

chain_with_retry = chain.with_retry(
    retry_if_exception_type=(TransientError,),
    stop_after_attempt=3,
)

---

Canonical Node Implementation

Follow this template for every graph node:

async def analyze_requirements(state: PlanGenerationState) -> dict:
    """Analyze requirements node with proper error handling.

    Args:
        state: Current workflow state containing 'description'.

    Returns:
        Dict with 'analysis' key on success, or 'error' key on failure.
    """
    try:
        prompt = ChatPromptTemplate.from_template(
            "Analyze these requirements and identify key constraints: {requirements}"
        )
        chain = prompt | llm | StrOutputParser()
        analysis = await chain.ainvoke({"requirements": state["description"]})
        return {"analysis": analysis}
    except Exception as e:
        return {"error": f"Analysis failed: {str(e)}"}

Mandatory elements of every node:

1. async def (async required for graph nodes)
2. Type-annotated state parameter using the workflow's TypedDict
3. Return type -> dict (partial state update)
4. Docstring explaining the node's purpose and return values
5. try/except with error captured in {"error": ...} dict
6. Never let exceptions propagate out of a node — always capture them

---

Testing LangChain/LangGraph Code

Mock Providers

from langchain_community.llms.fake import FakeListLLM

# Deterministic, no API keys, no cost:
mock_llm = FakeListLLM(responses=[
    "Analysis: The requirements describe a data processing pipeline.",
    "Plan: Step 1 — validate input. Step 2 — transform. Step 3 — output.",
])

# Inject into your service:
service = PlanService(llm=mock_llm)

Never use real LLM APIs in unit tests (features/). Only real LLMs in e2e tests (robot/ — nox -s e2e_tests).

Node State Testing

# Test each node independently:
result = await analyze_requirements({
    "description": "Build a CSV export feature",
    "analysis": "",
    "plan": "",
    "error": "",
    "is_complete": False,
})
assert "analysis" in result
assert result.get("error", "") == ""

Complete Workflow Testing

# Test the full graph with FakeListLLM:
app = create_plan_workflow(llm=mock_llm)
result = await app.ainvoke({"description": "test requirements"})
assert result["is_complete"] is True
assert result["error"] == ""

Streaming Testing

# Test both event emission and final result:
events = []
async for event in app.astream(initial_state):
    events.append(event)

assert len(events) > 0
assert events[-1]["is_complete"] is True

Memory Testing

# Verify conversation history accumulates correctly:
config = {"configurable": {"thread_id": "test-thread"}}
await app.ainvoke(state_1, config=config)
await app.ainvoke(state_2, config=config)  # same thread

history = memory.get(config)
assert len(history.messages) == 4  # 2 turns × 2 messages each