स्टेट मशीन और एजेंट: LangGraph के साथ विश्वसनीय वर्कफ्लो बनाना

messages = []
while True:
    response = claude.messages.create(model=..., messages=messages)
    if response.stop_reason == "end_turn":
        break
    for tool_call in get_tool_calls(response):
        result = execute_tool(tool_call)
        messages.append(tool_result(tool_call.id, result))

from typing import TypedDict

class ResearchState(TypedDict):
    query: str
    research_notes: str
    draft: str
    review_feedback: str
    is_approved: bool

import anthropic

client = anthropic.Anthropic()

def research_node(state: ResearchState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"इस विषय पर गहन शोध करें: {state['query']}"
        }]
    )
    return {"research_notes": response.content[0].text}

def draft_node(state: ResearchState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"इन नोट्स के आधार पर एक ड्राफ्ट लिखें:\n{state['research_notes']}"
        }]
    )
    return {"draft": response.content[0].text}

from typing import Literal

def route_after_check(state: ResearchState) -> Literal["human_review", "draft"]:
    if state.get("review_feedback"):
        return "human_review"
    return "draft"

workflow.add_conditional_edges(
    "research",
    route_after_check,
    {"human_review": "review_node", "draft": "draft_node"}
)

from langgraph.graph import StateGraph, END

workflow = StateGraph(ResearchState)
workflow.add_node("research", research_node)
workflow.add_node("draft", draft_node)
workflow.set_entry_point("research")
workflow.add_edge("research", "draft")
workflow.add_edge("draft", END)

graph = workflow.compile()
result = graph.invoke({"query": "MCP प्रोटोकॉल क्या है?"})
print(result["draft"])

class DocumentState(TypedDict):
    document: str
    key_terms: list[str]
    compliance_issues: list[str]
    summary: str
    human_feedback: str
    is_approved: bool

import json

def extract_terms_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": (
                "इस दस्तावेज़ से ठीक 5 मुख्य शब्द निकालें। "
                "उन्हें स्ट्रिंग्स के JSON ऐरे के रूप में लौटाएं।\n\n"
                f"दस्तावेज़: {state['document']}"
            )
        }]
    )
    try:
        raw = response.content[0].text.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1]
            if raw.startswith("json"):
                raw = raw[4:]
        terms = json.loads(raw.strip())
    except (json.JSONDecodeError, IndexError):
        terms = [t.strip() for t in response.content[0].text.split("\n") if t.strip()][:5]
    return {"key_terms": terms}


def check_compliance_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": (
                "इस दस्तावेज़ में अनुपालन समस्याओं की जांच करें। "
                "देखें: PII (नाम, आधार नंबर, ईमेल, फोन), "
                "गोपनीय चिह्न, असत्यापित दावे।\n\n"
                "समस्या विवरण का JSON ऐरे लौटाएं। "
                "कोई समस्या नहीं होने पर खाली ऐरे [] लौटाएं।\n\n"
                f"दस्तावेज़: {state['document']}"
            )
        }]
    )
    try:
        raw = response.content[0].text.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1]
            if raw.startswith("json"):
                raw = raw[4:]
        issues = json.loads(raw.strip())
    except (json.JSONDecodeError, IndexError):
        issues = []
    return {"compliance_issues": issues}


def human_review_node(state: DocumentState) -> dict:
    print("\n=== मानव समीक्षा आवश्यक ===")
    for issue in state["compliance_issues"]:
        print(f"  - {issue}")
    return {
        "human_feedback": "PII हटाने के बाद समीक्षा और अनुमोदित",
        "is_approved": True
    }


def summarize_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=256,
        messages=[{
            "role": "user",
            "content": f"इस दस्तावेज़ को ठीक दो वाक्यों में सारांशित करें:\n\n{state['document']}"
        }]
    )
    return {"summary": response.content[0].text, "is_approved": True}

def route_after_compliance(state: DocumentState) -> Literal["human_review", "summarize"]:
    if state.get("compliance_issues"):
        return "human_review"
    return "summarize"

workflow = StateGraph(DocumentState)
workflow.add_node("extract", extract_terms_node)
workflow.add_node("check", check_compliance_node)
workflow.add_node("review", human_review_node)
workflow.add_node("summarize", summarize_node)

workflow.set_entry_point("extract")
workflow.add_edge("extract", "check")
workflow.add_conditional_edges(
    "check", route_after_compliance,
    {"human_review": "review", "summarize": "summarize"}
)
workflow.add_edge("review", "summarize")
workflow.add_edge("summarize", END)

graph = workflow.compile()

result = graph.invoke({
    "document": "रोगी राम शर्मा (आधार: 1234-5678-9012) को उच्च रक्तचाप है।",
    "key_terms": [], "compliance_issues": [],
    "summary": "", "human_feedback": "", "is_approved": False,
})
print(result["summary"])

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
graph = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["review"]
)

thread_config = {"configurable": {"thread_id": "doc-review-001"}}
result = graph.invoke(initial_state, config=thread_config)

current_state = graph.get_state(thread_config)
print("समस्याएं:", current_state.values["compliance_issues"])

graph.update_state(
    thread_config,
    {"human_feedback": "मैन्युअल PII हटाने के बाद अनुमोदित", "is_approved": True}
)

final_result = graph.invoke(None, config=thread_config)

from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string("postgresql://...") as checkpointer:
    graph = workflow.compile(checkpointer=checkpointer)

import asyncio

async def batch_process(documents: list[str]) -> list[dict]:
    return await asyncio.gather(*[
        graph.ainvoke({"document": doc, "key_terms": [], "compliance_issues": [],
                       "summary": "", "human_feedback": "", "is_approved": False})
        for doc in documents
    ])

def test_extract_terms_node():
    state = {
        "document": "कृत्रिम बुद्धिमत्ता आधुनिक समाज को तेजी से बदल रही है।",
        "key_terms": [], "compliance_issues": [],
        "summary": "", "human_feedback": "", "is_approved": False,
    }
    result = extract_terms_node(state)
    assert "key_terms" in result
    assert isinstance(result["key_terms"], list)