AI-agent development has progressed through overlapping phases: prompt engineering, context engineering, tool use, autonomous loops, memory systems, and multi-agent coordination. A newer focus is graph engineering, which treats AI applications as explicitly designed workflows rather than a single autonomous agent.<\/p>\n
Graph engineering defines how agents, tools, deterministic functions, validators, data sources, and humans coordinate to complete tasks. It is broader than LangGraph, GraphRAG, or knowledge graphs. In this article, we examine graph engineering from an implementation perspective and build a reliable LangGraph workflow.<\/p>\n
What Is Graph Engineering?<\/strong><\/h2>\nGraph engineering is the practice of representing an AI application as an executable graph containing agents, tools, functions, policies, data systems, evaluators, and human decisions.<\/p>\n
A practical definition is:<\/p>\n
<\/p>\n
Graph engineering is the design of nodes, dependencies, state transitions, execution routes, validation gates, recovery paths, and control boundaries inside an agentic system.<\/p>\n
\n<\/p><\/blockquote>\n
Consider an AI system that researches a technical topic, writes a report, verifies its claims, and sends it to a client.<\/p>\n
A single-agent implementation might look like this:<\/p>\n
\u00a0<\/p>\n
\n![\"What](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2026\/07\/image-2-pa10g5.webp\")
<\/figure>\n<\/div>\nMost of those transitions are hidden inside the model\u2019s context. The model decides when to search, when enough evidence has been collected, whether the output is correct, and when the task is complete.<\/p>\n
A graph-engineered implementation makes those responsibilities explicit:<\/p>\n
\n![\"Workflow](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2026\/07\/image-3-pa10g5-scaled.webp\")
<\/figure>\n<\/div>\nHere, the graph defines which transitions are permitted. Individual agents can still reason autonomously within their nodes, but they do not control the entire system.<\/p>\n
Core Components of Graph Engineering<\/strong><\/h2>\n1. Nodes<\/strong><\/h3>\nA node is a bounded unit of execution.<\/p>\n
A node may contain:<\/p>\n
<\/p>\n- An LLM call<\/li>\n
\n
\n
\n
<\/p>\n
- A complete tool-using agent<\/li>\n
\n
\n
\n
<\/p>\n
- A Python function<\/li>\n
\n
\n
\n
<\/p>\n
- A retrieval operation<\/li>\n
\n
\n
\n
<\/p>\n
- A database query<\/li>\n
\n
\n
\n
<\/p>\n
- An API request<\/li>\n
\n
\n
\n
<\/p>\n
- A policy check<\/li>\n
\n
\n
\n
<\/p>\n
- A test suite<\/li>\n
\n
\n
\n
<\/p>\n
- A human approval request<\/li>\n
\n
\n
\n
<\/p>\n
- A subgraph<\/li>\n
\n<\/ul>\n
Not every node should be an AI agent.<\/p>\n
Known business rules should generally remain deterministic. An LLM is useful where semantic interpretation, generation, planning, or ambiguity is involved.<\/p>\n
For example, calculating whether an invoice exceeds an approval threshold does not require an LLM. Understanding whether an email represents a refund request may require one.<\/p>\n
2. Edges<\/strong><\/h3>\nEdges define which nodes can execute after another node.<\/p>\n
Common edge types include:<\/p>\n
<\/p>\n- Direct edges<\/li>\n
\n
\n
\n
<\/p>\n
- Conditional edges<\/li>\n
\n
\n
\n
<\/p>\n
- Parallel edges<\/li>\n
\n
\n
\n
<\/p>\n
- Looping edges<\/li>\n
\n
\n
\n
<\/p>\n
- Error edges<\/li>\n
\n
\n
\n
<\/p>\n
- Human-controlled edges<\/li>\n
\n
\n
\n
<\/p>\n
- Event-triggered edges<\/li>\n
\n<\/ul>\n
An edge represents a dependency or control rule.<\/p>\n
For example:<\/p>\n
\n![\"Edges](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2026\/07\/image-4-pa10g5.webp\")
<\/figure>\n<\/div>\nThe routing condition may be implemented through deterministic Python logic or an LLM classifier.<\/p>\n
3. State<\/strong><\/h3>\nState is the shared record carried through the graph.<\/p>\n
It may contain:<\/p>\n
class WorkflowState(TypedDict):
\n user_request: str
\n task_plan: list[str]
\n retrieved_evidence: list[str]
\n draft: str
\n validation_result: dict
\n retry_count: int
\n approval_status: str<\/code><\/pre>\nTyped state makes the inputs and outputs of nodes visible. It also reduces the need to pass a complete conversation transcript to every agent.<\/p>\n
LangGraph uses stateful graphs to combine deterministic steps with LLM-driven steps. It also provides persistence, streaming, human-in-the-loop controls, and support for long-running execution.<\/p>\n
4. State Reducers<\/strong><\/h3>\nParallel nodes may update the same state field.<\/p>\n
Suppose three research agents return evidence simultaneously:<\/p>\n
{
\n \"evidence\": [\"Source A\"]
\n}
\n{
\n \"evidence\": [\"Source B\"]
\n}
\n{
\n \"evidence\": [\"Source C\"]
\n}<\/code><\/pre>\nThe graph needs a rule for combining these updates.<\/p>\n
A reducer may append lists, merge dictionaries, select the latest value, or apply a custom conflict-resolution policy.<\/p>\n
Without a clear reducer, parallel updates can overwrite one another or create inconsistent state.<\/p>\n
5. Routes and Guard Conditions<\/strong><\/h3>\nA route determines which edge should run.<\/p>\n
A guard condition checks whether a transition is allowed.<\/p>\n
def route_after_review(state):
\n if state[\"grounding_score\"] <\/code><\/pre>\nHard constraints should not be hidden inside prompts. They should be enforced in routing code whenever possible.<\/p>\n
6. Checkpoints<\/strong><\/h3>\nA checkpoint stores a snapshot of the graph state.<\/p>\n
Checkpoints allow a workflow to:<\/p>\n
<\/p>\n- Resume after interruption<\/li>\n
\n
\n
\n
<\/p>\n
- Recover from failure<\/li>\n
\n
\n
\n
<\/p>\n
- Wait for human feedback<\/li>\n
\n
\n
\n
<\/p>\n
- Inspect previous states<\/li>\n
\n
\n
\n
<\/p>\n
- Replay a workflow<\/li>\n
\n
\n
\n
<\/p>\n
- Support long-running tasks<\/li>\n
\n<\/ul>\n
LangGraph separates thread-level checkpoints from long-term stores. Checkpointers preserve graph state for a specific execution thread, while stores keep application data across threads.<\/p>\n
7. Interrupts<\/strong><\/h3>\nAn interrupt pauses the graph and requests external input.<\/p>\n
Common uses include:<\/p>\n
<\/p>\n- Approval before sending an email<\/li>\n
\n
\n
\n
<\/p>\n
- Review before publishing content<\/li>\n
\n
\n
\n
<\/p>\n
- Confirmation before issuing a refund<\/li>\n
\n
\n
\n
<\/p>\n
- Editing generated parameters<\/li>\n
\n
\n
\n
<\/p>\n
- Providing missing information<\/li>\n
\n<\/ul>\n
LangGraph interrupts save the current state and allow execution to resume later using the same thread identifier. The documentation also recommends ensuring that side effects before an interrupt are idempotent.<\/p>\n
Important Agent Graph Patterns<\/strong><\/h2>\nLangGraph and Anthropic describe several recurring orchestration patterns for agentic applications.<\/p>\n
Prompt Chaining<\/strong><\/h3>\nEach node processes the output of the previous node.<\/p>\n
\u00a0<\/p>\n
\n![\"Prompt](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2026\/07\/image-5-pa10g5.webp\")
<\/figure>\n<\/div>\nUse it when the task can be divided into fixed, verifiable stages.<\/p>\n
Routing<\/strong><\/h3>\nA router sends the request to a specialized branch.<\/p>\n
\u00a0<\/p>\n
\n![\"routing](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2026\/07\/image-6-pa10g5.webp\")
<\/figure>\n<\/div>\nUse deterministic routing when categories are exact. Use model-based routing when classification requires semantic interpretation.<\/p>\n
Parallelization<\/strong><\/h3>\nIndependent tasks execute concurrently.<\/p>\n
\n![\"parallelization](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2026\/07\/image-7-pa10g5.webp\")
<\/figure>\n<\/div>\nParallelization can reduce latency and allow different agents to examine separate dimensions of a problem.<\/p>\n
However, only independent tasks should run in parallel. Creating parallel branches that secretly depend on one another produces incomplete or inconsistent results.<\/p>\n
Orchestrator-Worker<\/strong><\/h3>\nAn orchestrator decomposes a task and delegates parts to workers.<\/p>\n
\u00a0<\/p>\n
\n![\"Orchestrator-Worker](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2026\/07\/image-8-pa10g5.webp\")
<\/figure>\n<\/div>\nThis pattern is useful when the number and type of subtasks cannot be known before the request arrives.<\/p>\n
The orchestrator should primarily plan, assign, and integrate. If it directly performs every tool call, the architecture becomes another monolithic agent.<\/p>\n
Evaluator-Optimizer<\/strong><\/h3>\nOne component generates an artifact while another evaluates it.<\/p>\n
\n![\"Evaluator-Optimizer](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2026\/07\/image-9-pa10g5.webp\")
<\/figure>\n<\/div>\nThis pattern works best when the evaluation criteria are clear and revision produces measurable improvement.<\/p>\n
Human-in-the-Loop<\/strong><\/h3>\nA human reviews the workflow before a consequential action.<\/p>\n
\n![\"Human-in-the-Loop](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2026\/07\/image-10-pa10g5.webp\")
<\/figure>\n<\/div>\nHuman review should be risk-based. Requiring approval for every harmless step makes the system slow without improving safety.<\/p>\n
Hands-On: Building a Graph-Engineered Research Workflow<\/strong><\/h2>\nWe will build a graph that:<\/p>\n
<\/p>\n- Plans the task<\/li>\n
\n
\n
\n
<\/p>\n
- Collects evidence<\/li>\n
\n
\n
\n
<\/p>\n
- Writes a draft<\/li>\n
\n
\n
\n
<\/p>\n
- Evaluates the draft<\/li>\n
\n
\n
\n
<\/p>\n
- Revises weak drafts<\/li>\n
\n
\n
\n
<\/p>\n
- Requests human approval<\/li>\n
\n
\n
\n
<\/p>\n
- Finalizes the output<\/li>\n
\n<\/ol>\n
Install the Dependencies<\/strong><\/h3>\npip install -U langgraph langchain langchain-openai<\/code><\/pre>\nInstall the integration package for your chosen model provider separately.<\/p>\n
Set a supported model in the environment:<\/p>\n
model_name =\"openai:gpt-4.1-mini\"<\/code><\/pre>\nDefine the State and Model<\/strong><\/h3>\nimport os
\nfrom typing import Literal, TypedDict
\n
\nfrom pydantic import BaseModel, Field
\nfrom langchain.chat_models import init_chat_model
\nfrom langgraph.checkpoint.memory import InMemorySaver
\nfrom langgraph.graph import END, START, StateGraph
\nfrom langgraph.types import Command, interrupt
\n
\nfrom google.colab import userdata
\nos.environ['OPENAI_API_KEY'] = userdata.get('OPENAI_KEY')
\n
\nmodel = init_chat_model(model_name)
\n
\n
\nclass ResearchState(TypedDict, total=False):
\n topic: str
\n plan: str
\n evidence: str
\n draft: str
\n feedback: str
\n evaluator_approved: bool
\n human_approved: bool
\n revision_count: int
\n
\n
\nclass ReviewResult(BaseModel):
\n approved: bool = Field(
\n description=\"Whether the draft is accurate and well grounded.\"
\n )
\n feedback: str = Field(
\n description=\"Specific corrections required before approval.\"
\n )
\n
\n
\nreview_model = model.with_structured_output(ReviewResult)<\/code><\/pre>\nThe state is the interface shared by all nodes. Each node reads only the values it needs and returns only the fields it o wns.<\/p>\n
Create the Nodes<\/strong><\/h3>\ndef planner_node(state: ResearchState) -> ResearchState:
\n response = model.invoke(
\n f\"\"\"
\n Create a concise research plan for the following topic:
\n
\n {state['topic']}
\n
\n Include the questions that must be answered and the evidence needed.
\n \"\"\"
\n )
\n
\n return {
\n \"plan\": response.content,
\n \"revision_count\": 0,
\n }
\n
\n
\ndef researcher_node(state: ResearchState) -> ResearchState:
\n response = model.invoke(
\n f\"\"\"
\n Produce a grounded research brief for this topic:
\n
\n {state['topic']}
\n
\n Follow this plan:
\n
\n {state['plan']}
\n
\n Clearly separate verified information, assumptions, and open questions.
\n \"\"\"
\n )
\n
\n return {\"evidence\": response.content}
\n
\n
\ndef writer_node(state: ResearchState) -> ResearchState:
\n response = model.invoke(
\n f\"\"\"
\n Write a professional technical article using only the evidence below.
\n
\n Topic:
\n {state['topic']}
\n
\n Evidence:
\n {state['evidence']}
\n
\n Include an introduction, architecture explanation, implementation
\n considerations, limitations, and conclusion.
\n \"\"\"
\n )
\n
\n return {\"draft\": response.content}
\n
\n
\ndef evaluator_node(state: ResearchState) -> ResearchState:
\n review = review_model.invoke(
\n f\"\"\"
\n Evaluate the draft against the available evidence.
\n
\n Evidence:
\n {state['evidence']}
\n
\n Draft:
\n {state['draft']}
\n
\n Check technical accuracy, grounding, completeness, and clarity.
\n \"\"\"
\n )
\n
\n return {
\n \"evaluator_approved\": review.approved,
\n \"feedback\": review.feedback,
\n }
\n
\n
\ndef revision_node(state: ResearchState) -> ResearchState:
\n response = model.invoke(
\n f\"\"\"
\n Revise the following technical article.
\n
\n Draft:
\n {state['draft']}
\n
\n Reviewer feedback:
\n {state['feedback']}
\n
\n Preserve correct sections and fix only the identified weaknesses.
\n \"\"\"
\n )
\n
\n return {
\n \"draft\": response.content,
\n \"revision_count\": state.get(\"revision_count\", 0) + 1,
\n }
\n
\n
\ndef human_review_node(state: ResearchState) -> ResearchState:
\n decision = interrupt(
\n {
\n \"message\": \"Review this article before finalization.\",
\n \"draft\": state[\"draft\"],
\n \"automated_feedback\": state.get(\"feedback\", \"\"),
\n \"allowed_actions\": [\"approve\", \"reject\"],
\n }
\n )
\n
\n return {
\n \"human_approved\": decision.get(\"action\") == \"approve\",
\n \"feedback\": decision.get(
\n \"feedback\",
\n state.get(\"feedback\", \"\"),
\n ),
\n }
\n
\n
\ndef finalize_node(state: ResearchState) -> ResearchState:
\n return {\"human_approved\": True}<\/code><\/pre>\nDefine Conditional Routes<\/strong><\/h3>\ndef route_after_evaluation(
\n state: ResearchState,
\n) -> Literal[\"revise\", \"human_review\"]:
\n if state.get(\"evaluator_approved\"):
\n return \"human_review\"
\n
\n if state.get(\"revision_count\", 0) >= 2:
\n return \"human_review\"
\n
\n return \"revise\"
\n
\n
\ndef route_after_human_review(
\n state: ResearchState,
\n) -> Literal[\"finalize\", \"revise\"]:
\n if state.get(\"human_approved\"):
\n return \"finalize\"
\n
\n return \"revise\"<\/code><\/pre>\nThe evaluator does not control the graph directly. It updates the state. The routing function reads that state and selects an allowed edge.<\/p>\n
The revision limit prevents an unbounded evaluator-optimizer loop.<\/p>\n
Construct the Graph<\/strong><\/h3>\nbuilder = StateGraph(ResearchState)
\n
\nbuilder.add_node(\"planner\", planner_node)
\nbuilder.add_node(\"researcher\", researcher_node)
\nbuilder.add_node(\"writer\", writer_node)
\nbuilder.add_node(\"evaluator\", evaluator_node)
\nbuilder.add_node(\"revise\", revision_node)
\nbuilder.add_node(\"human_review\", human_review_node)
\nbuilder.add_node(\"finalize\", finalize_node)
\n
\nbuilder.add_edge(START, \"planner\")
\nbuilder.add_edge(\"planner\", \"researcher\")
\nbuilder.add_edge(\"researcher\", \"writer\")
\nbuilder.add_edge(\"writer\", \"evaluator\")
\n
\nbuilder.add_conditional_edges(
\n \"evaluator\",
\n route_after_evaluation,
\n {
\n \"revise\": \"revise\",
\n \"human_review\": \"human_review\",
\n },
\n)
\n
\nbuilder.add_edge(\"revise\", \"evaluator\")
\n
\nbuilder.add_conditional_edges(
\n \"human_review\",
\n route_after_human_review,
\n {
\n \"finalize\": \"finalize\",
\n \"revise\": \"revise\",
\n },
\n)
\n
\nbuilder.add_edge(\"finalize\", END)
\n
\ncheckpointer = InMemorySaver()
\ngraph = builder.compile(checkpointer=checkpointer)<\/code><\/pre>\n\n![\"Construct](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2026\/07\/image-11-pa10g5.webp\")
<\/figure>\n<\/div>\nRun the Graph<\/strong><\/h3>\nconfig = {
\n \"configurable\": {
\n \"thread_id\": \"graph-engineering-article-001\"
\n }
\n}
\n
\nresult = graph.invoke(
\n {
\n \"topic\": \"Graph engineering for reliable AI agents\",
\n \"revision_count\": 0,
\n },
\n config=config,
\n)
\n
\nif \"__interrupt__\" in result:
\n print(\"The workflow is waiting for human approval.\")<\/code><\/pre>\nResume After Approval<\/strong><\/h3>\nfinal_state = graph.invoke(
\n Command(
\n resume={
\n \"action\": \"approve\",
\n \"feedback\": \"Approved for publication.\",
\n }
\n ),
\n config=config,
\n)
\n
\nprint(final_state[\"draft\"])<\/code><\/pre>\n\n![\"The](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2026\/07\/image-12-pa10g5.webp\")
<\/figure>\n<\/div>\nThe same thread_id<\/code> is required because it identifies the stored execution state.<\/p>\nInMemorySaver<\/code> is suitable for demonstration, but it loses all checkpoints when the process restarts. Production systems should use durable persistence backed by a database.<\/p>\nProduction Requirements That Diagrams Often Miss<\/strong><\/h2>\nA graph that runs in a notebook is not automatically production-ready.<\/p>\n
Node Contracts<\/strong><\/h3>\nEvery node should define:<\/p>\n
<\/p>\n- Required inputs<\/li>\n
\n
\n
\n
<\/p>\n
- Produced outputs<\/li>\n
\n
\n
\n
<\/p>\n
- Allowed tools<\/li>\n
\n
\n
\n
<\/p>\n
- Timeout<\/li>\n
\n
\n
\n
<\/p>\n
- Retry policy<\/li>\n
\n
\n
\n
<\/p>\n
- Side effects<\/li>\n
\n
\n
\n
<\/p>\n
- Failure categories<\/li>\n
\n
\n
\n
<\/p>\n
- Validation rules<\/li>\n
\n
\n
\n
<\/p>\n
- Ownership<\/li>\n
\n<\/ul>\n
A node that accepts arbitrary state and returns unstructured text becomes difficult to test.<\/p>\n
Idempotency<\/strong><\/h3>\nA retry should not repeat an irreversible action.<\/p>\n
For example, retrying a payment node must not charge the customer twice.<\/p>\n
Use:<\/p>\n
<\/p>\n- Idempotency keys<\/li>\n
\n
\n
\n
<\/p>\n
- Transaction identifiers<\/li>\n
\n
\n
\n
<\/p>\n
- Deduplication checks<\/li>\n
\n
\n
\n
<\/p>\n
- Operation logs<\/li>\n
\n
\n
\n
<\/p>\n
- Database constraints<\/li>\n
\n<\/ul>\n
Error Classification<\/strong><\/h3>\nNot every failure should be retried.<\/p>\n
Temporary network failure -> RetryA generic retry loop can increase cost without fixing the underlying issue.<\/p>\n
Context Isolation<\/strong><\/h3>\nDo not give every node the complete graph state.<\/p>\n
<\/p>\n- A writer may need evidence and an outline.<\/li>\n
\n
\n
\n
<\/p>\n
- A security reviewer may need code and deployment configuration.<\/li>\n
\n
\n
\n
<\/p>\n
- A publication node may need only the approved artifact and destination.<\/li>\n
\n<\/ol>\n
Context isolation reduces token usage, accidental data exposure, and distraction from irrelevant history.<\/p>\n
Observability<\/strong><\/h3>\nTrace at least:<\/p>\n
<\/p>\n- Node start and completion<\/li>\n
\n
\n
\n
<\/p>\n
- Selected route<\/li>\n
\n
\n
\n
<\/p>\n
- State fields changed<\/li>\n
\n
\n
\n
<\/p>\n
- Tool calls<\/li>\n
\n
\n
\n
<\/p>\n
- Model and prompt version<\/li>\n
\n
\n
\n
<\/p>\n
- Token consumption<\/li>\n
\n
\n
\n
<\/p>\n
- Latency<\/li>\n
\n
\n
\n
<\/p>\n
- Retry count<\/li>\n
\n
\n
\n
<\/p>\n
- Validation results<\/li>\n
\n
\n
\n
<\/p>\n
- Human decisions<\/li>\n
\n
\n
\n
<\/p>\n
- Final outcome<\/li>\n
\n<\/ul>\n
Microsoft Agent Framework also separates dynamic agents from explicitly controlled workflows. Its workflow system supports graph-based execution, typed message routing, conditional paths, parallel processing, checkpoints, and human-in-the-loop interactions.<\/p>\n
Framework Options<\/strong><\/h2>\nLangGraph<\/strong><\/h3>\nBest suited for teams that need low-level control over state, routes, persistence, subgraphs, interrupts, and mixed deterministic-agentic execution.<\/p>\n
Google ADK<\/strong><\/h3>\nUseful for teams building in the Google ecosystem. It supports multi-agent workflows, template-based sequential, parallel, and loop execution, and newer graph-oriented workflows.<\/p>\n
Microsoft Agent Framework<\/strong><\/h3>\nSuitable for Python, .NET, and Go teams that require typed workflows, graph routing, checkpointing, human interaction, and integration with enterprise systems.<\/p>\n
Plain Python and Existing Workflow Engines<\/strong><\/h3>\nA specialized agent framework may be unnecessary when most of the workflow is deterministic.<\/p>\n
Python functions, queues, databases, schedulers, and state-machine libraries can coordinate a small number of LLM-powered steps effectively.<\/p>\n
Limitations<\/strong><\/h2>\nGraph engineering also introduces:<\/p>\n
<\/p>\n- Additional infrastructure<\/li>\n
\n
\n
\n
<\/p>\n
- More state-management complexity<\/li>\n
\n
\n
\n
<\/p>\n
- Higher testing requirements<\/li>\n
\n
\n
\n
<\/p>\n
- Synchronization challenges<\/li>\n
\n
\n
\n
<\/p>\n
- Increased model cost when many agents are used<\/li>\n
\n
\n
\n
<\/p>\n
- More difficult versioning and migrations<\/li>\n
\n
\n
\n
<\/p>\n
- Potential latency at parallel join points<\/li>\n
\n
\n
\n
<\/p>\n
- Overengineering for simple tasks<\/li>\n
\n<\/ul>\n
A graph is useful when its structure makes the system safer, faster, easier to evaluate, or easier to maintain.<\/p>\n
It is not valuable merely because it contains more boxes.<\/p>\n
Conclusion<\/strong><\/h2>\nGraph engineering is not a replacement for prompt engineering, context engineering, harness engineering, or loop engineering.<\/p>\n
It is the layer that coordinates them.<\/p>\n
<\/p>\n- Prompts control individual model calls.<\/li>\n
\n
\n
\n
<\/p>\n
- Context engineering controls what each model sees.<\/li>\n
\n
\n
\n
<\/p>\n
- Agent loops control how an agent reasons and uses tools.<\/li>\n
\n
\n
\n
<\/p>\n
- Graph engineering controls how multiple agents, loops, functions, validators, tools, and humans work together.<\/li>\n
\n<\/ul>\n
The broader lesson is simple:<\/p>\n
<\/p>\n- Do not begin by asking how many agents the system needs.<\/li>\n
\n
\n
\n
<\/p>\n
- Begin by identifying the work, dependencies, decision boundaries, parallel opportunities, validation requirements, failure paths, and human responsibilities.<\/li>\n
\n
\n
\n
<\/p>\n
- The agents fill the nodes.<\/li>\n
\n
\n
\n
<\/p>\n
- The engineering lives in the edges.<\/li>\n
\n<\/ul>\n
Frequently Asked Questions<\/strong><\/h2>\n\nIs graph engineering the same as LangGraph?<\/strong><\/p>\nNo. LangGraph is one framework for implementing graph-based agent workflows. Graph engineering is the broader architectural practice. Are knowledge graphs required? No. Workflow graphs can operate without a knowledge graph. A knowledge graph is useful when the system needs to represent and traverse relationships between entities.<\/p>\n<\/div>\n
Can a graph contain loops?<\/strong><\/p>\nYes. Revision cycles, tool-calling agents, retries, and evaluator-optimizer patterns are loops inside a larger graph.<\/p>\n<\/div>\n
Does every node need an LLM?<\/strong><\/p>\nNo. Most production graphs should combine deterministic functions, tools, policies, and selected LLM-powered nodes.<\/p>\n<\/div>\n
Can one agent be enough?<\/strong><\/p>\nYes. A single agent is often preferable for low-risk, open-ended tasks with a limited toolset and simple recovery requirements.<\/p>\n<\/div>\n<\/div>\n
<\/p>\n
Graph engineering is the design of nodes, dependencies, state transitions, execution routes, validation gates, recovery paths, and control boundaries inside an agentic system.<\/p>\n
\n<\/p><\/blockquote>\n
Consider an AI system that researches a technical topic, writes a report, verifies its claims, and sends it to a client.<\/p>\n
A single-agent implementation might look like this:<\/p>\n
\u00a0<\/p>\n
<\/figure>\n<\/div>\nMost of those transitions are hidden inside the model\u2019s context. The model decides when to search, when enough evidence has been collected, whether the output is correct, and when the task is complete.<\/p>\n
A graph-engineered implementation makes those responsibilities explicit:<\/p>\n
<\/figure>\n<\/div>\nHere, the graph defines which transitions are permitted. Individual agents can still reason autonomously within their nodes, but they do not control the entire system.<\/p>\n
Core Components of Graph Engineering<\/strong><\/h2>\n1. Nodes<\/strong><\/h3>\nA node is a bounded unit of execution.<\/p>\n
A node may contain:<\/p>\n
<\/p>\n- An LLM call<\/li>\n
\n
\n
\n
<\/p>\n
- A complete tool-using agent<\/li>\n
\n
\n
\n
<\/p>\n
- A Python function<\/li>\n
\n
\n
\n
<\/p>\n
- A retrieval operation<\/li>\n
\n
\n
\n
<\/p>\n
- A database query<\/li>\n
\n
\n
\n
<\/p>\n
- An API request<\/li>\n
\n
\n
\n
<\/p>\n
- A policy check<\/li>\n
\n
\n
\n
<\/p>\n
- A test suite<\/li>\n
\n
\n
\n
<\/p>\n
- A human approval request<\/li>\n
\n
\n
\n
<\/p>\n
- A subgraph<\/li>\n
\n<\/ul>\n
Not every node should be an AI agent.<\/p>\n
Known business rules should generally remain deterministic. An LLM is useful where semantic interpretation, generation, planning, or ambiguity is involved.<\/p>\n
For example, calculating whether an invoice exceeds an approval threshold does not require an LLM. Understanding whether an email represents a refund request may require one.<\/p>\n
2. Edges<\/strong><\/h3>\nEdges define which nodes can execute after another node.<\/p>\n
Common edge types include:<\/p>\n
<\/p>\n- Direct edges<\/li>\n
\n
\n
\n
<\/p>\n
- Conditional edges<\/li>\n
\n
\n
\n
<\/p>\n
- Parallel edges<\/li>\n
\n
\n
\n
<\/p>\n
- Looping edges<\/li>\n
\n
\n
\n
<\/p>\n
- Error edges<\/li>\n
\n
\n
\n
<\/p>\n
- Human-controlled edges<\/li>\n
\n
\n
\n
<\/p>\n
- Event-triggered edges<\/li>\n
\n<\/ul>\n
An edge represents a dependency or control rule.<\/p>\n
For example:<\/p>\n
\n<\/figure>\n<\/div>\nThe routing condition may be implemented through deterministic Python logic or an LLM classifier.<\/p>\n
3. State<\/strong><\/h3>\nState is the shared record carried through the graph.<\/p>\n
It may contain:<\/p>\n
class WorkflowState(TypedDict):
\n user_request: str
\n task_plan: list[str]
\n retrieved_evidence: list[str]
\n draft: str
\n validation_result: dict
\n retry_count: int
\n approval_status: str<\/code><\/pre>\nTyped state makes the inputs and outputs of nodes visible. It also reduces the need to pass a complete conversation transcript to every agent.<\/p>\n
LangGraph uses stateful graphs to combine deterministic steps with LLM-driven steps. It also provides persistence, streaming, human-in-the-loop controls, and support for long-running execution.<\/p>\n
4. State Reducers<\/strong><\/h3>\nParallel nodes may update the same state field.<\/p>\n
Suppose three research agents return evidence simultaneously:<\/p>\n
{
\n \"evidence\": [\"Source A\"]
\n}
\n{
\n \"evidence\": [\"Source B\"]
\n}
\n{
\n \"evidence\": [\"Source C\"]
\n}<\/code><\/pre>\nThe graph needs a rule for combining these updates.<\/p>\n
A reducer may append lists, merge dictionaries, select the latest value, or apply a custom conflict-resolution policy.<\/p>\n
Without a clear reducer, parallel updates can overwrite one another or create inconsistent state.<\/p>\n
5. Routes and Guard Conditions<\/strong><\/h3>\nA route determines which edge should run.<\/p>\n
A guard condition checks whether a transition is allowed.<\/p>\n
def route_after_review(state):
\n if state[\"grounding_score\"] <\/code><\/pre>\nHard constraints should not be hidden inside prompts. They should be enforced in routing code whenever possible.<\/p>\n
6. Checkpoints<\/strong><\/h3>\nA checkpoint stores a snapshot of the graph state.<\/p>\n
Checkpoints allow a workflow to:<\/p>\n
<\/p>\n- Resume after interruption<\/li>\n
\n
\n
\n
<\/p>\n
- Recover from failure<\/li>\n
\n
\n
\n
<\/p>\n
- Wait for human feedback<\/li>\n
\n
\n
\n
<\/p>\n
- Inspect previous states<\/li>\n
\n
\n
\n
<\/p>\n
- Replay a workflow<\/li>\n
\n
\n
\n
<\/p>\n
- Support long-running tasks<\/li>\n
\n<\/ul>\n
LangGraph separates thread-level checkpoints from long-term stores. Checkpointers preserve graph state for a specific execution thread, while stores keep application data across threads.<\/p>\n
7. Interrupts<\/strong><\/h3>\nAn interrupt pauses the graph and requests external input.<\/p>\n
Common uses include:<\/p>\n
<\/p>\n- Approval before sending an email<\/li>\n
\n
\n
\n
<\/p>\n
- Review before publishing content<\/li>\n
\n
\n
\n
<\/p>\n
- Confirmation before issuing a refund<\/li>\n
\n
\n
\n
<\/p>\n
- Editing generated parameters<\/li>\n
\n
\n
\n
<\/p>\n
- Providing missing information<\/li>\n
\n<\/ul>\n
LangGraph interrupts save the current state and allow execution to resume later using the same thread identifier. The documentation also recommends ensuring that side effects before an interrupt are idempotent.<\/p>\n
Important Agent Graph Patterns<\/strong><\/h2>\nLangGraph and Anthropic describe several recurring orchestration patterns for agentic applications.<\/p>\n
Prompt Chaining<\/strong><\/h3>\nEach node processes the output of the previous node.<\/p>\n
\u00a0<\/p>\n
\n<\/figure>\n<\/div>\nUse it when the task can be divided into fixed, verifiable stages.<\/p>\n
Routing<\/strong><\/h3>\nA router sends the request to a specialized branch.<\/p>\n
\u00a0<\/p>\n
\n<\/figure>\n<\/div>\nUse deterministic routing when categories are exact. Use model-based routing when classification requires semantic interpretation.<\/p>\n
Parallelization<\/strong><\/h3>\nIndependent tasks execute concurrently.<\/p>\n
\n<\/figure>\n<\/div>\nParallelization can reduce latency and allow different agents to examine separate dimensions of a problem.<\/p>\n
However, only independent tasks should run in parallel. Creating parallel branches that secretly depend on one another produces incomplete or inconsistent results.<\/p>\n
Orchestrator-Worker<\/strong><\/h3>\nAn orchestrator decomposes a task and delegates parts to workers.<\/p>\n
\u00a0<\/p>\n
\n<\/figure>\n<\/div>\nThis pattern is useful when the number and type of subtasks cannot be known before the request arrives.<\/p>\n
The orchestrator should primarily plan, assign, and integrate. If it directly performs every tool call, the architecture becomes another monolithic agent.<\/p>\n
Evaluator-Optimizer<\/strong><\/h3>\nOne component generates an artifact while another evaluates it.<\/p>\n
\n<\/figure>\n<\/div>\nThis pattern works best when the evaluation criteria are clear and revision produces measurable improvement.<\/p>\n
Human-in-the-Loop<\/strong><\/h3>\nA human reviews the workflow before a consequential action.<\/p>\n
\n<\/figure>\n<\/div>\nHuman review should be risk-based. Requiring approval for every harmless step makes the system slow without improving safety.<\/p>\n
Hands-On: Building a Graph-Engineered Research Workflow<\/strong><\/h2>\nWe will build a graph that:<\/p>\n
<\/p>\n- Plans the task<\/li>\n
\n
\n
\n
<\/p>\n
- Collects evidence<\/li>\n
\n
\n
\n
<\/p>\n
- Writes a draft<\/li>\n
\n
\n
\n
<\/p>\n
- Evaluates the draft<\/li>\n
\n
\n
\n
<\/p>\n
- Revises weak drafts<\/li>\n
\n
\n
\n
<\/p>\n
- Requests human approval<\/li>\n
\n
\n
\n
<\/p>\n
- Finalizes the output<\/li>\n
\n<\/ol>\n
Install the Dependencies<\/strong><\/h3>\npip install -U langgraph langchain langchain-openai<\/code><\/pre>\nInstall the integration package for your chosen model provider separately.<\/p>\n
Set a supported model in the environment:<\/p>\n
model_name =\"openai:gpt-4.1-mini\"<\/code><\/pre>\nDefine the State and Model<\/strong><\/h3>\nimport os
\nfrom typing import Literal, TypedDict
\n
\nfrom pydantic import BaseModel, Field
\nfrom langchain.chat_models import init_chat_model
\nfrom langgraph.checkpoint.memory import InMemorySaver
\nfrom langgraph.graph import END, START, StateGraph
\nfrom langgraph.types import Command, interrupt
\n
\nfrom google.colab import userdata
\nos.environ['OPENAI_API_KEY'] = userdata.get('OPENAI_KEY')
\n
\nmodel = init_chat_model(model_name)
\n
\n
\nclass ResearchState(TypedDict, total=False):
\n topic: str
\n plan: str
\n evidence: str
\n draft: str
\n feedback: str
\n evaluator_approved: bool
\n human_approved: bool
\n revision_count: int
\n
\n
\nclass ReviewResult(BaseModel):
\n approved: bool = Field(
\n description=\"Whether the draft is accurate and well grounded.\"
\n )
\n feedback: str = Field(
\n description=\"Specific corrections required before approval.\"
\n )
\n
\n
\nreview_model = model.with_structured_output(ReviewResult)<\/code><\/pre>\nThe state is the interface shared by all nodes. Each node reads only the values it needs and returns only the fields it o wns.<\/p>\n
Create the Nodes<\/strong><\/h3>\ndef planner_node(state: ResearchState) -> ResearchState:
\n response = model.invoke(
\n f\"\"\"
\n Create a concise research plan for the following topic:
\n
\n {state['topic']}
\n
\n Include the questions that must be answered and the evidence needed.
\n \"\"\"
\n )
\n
\n return {
\n \"plan\": response.content,
\n \"revision_count\": 0,
\n }
\n
\n
\ndef researcher_node(state: ResearchState) -> ResearchState:
\n response = model.invoke(
\n f\"\"\"
\n Produce a grounded research brief for this topic:
\n
\n {state['topic']}
\n
\n Follow this plan:
\n
\n {state['plan']}
\n
\n Clearly separate verified information, assumptions, and open questions.
\n \"\"\"
\n )
\n
\n return {\"evidence\": response.content}
\n
\n
\ndef writer_node(state: ResearchState) -> ResearchState:
\n response = model.invoke(
\n f\"\"\"
\n Write a professional technical article using only the evidence below.
\n
\n Topic:
\n {state['topic']}
\n
\n Evidence:
\n {state['evidence']}
\n
\n Include an introduction, architecture explanation, implementation
\n considerations, limitations, and conclusion.
\n \"\"\"
\n )
\n
\n return {\"draft\": response.content}
\n
\n
\ndef evaluator_node(state: ResearchState) -> ResearchState:
\n review = review_model.invoke(
\n f\"\"\"
\n Evaluate the draft against the available evidence.
\n
\n Evidence:
\n {state['evidence']}
\n
\n Draft:
\n {state['draft']}
\n
\n Check technical accuracy, grounding, completeness, and clarity.
\n \"\"\"
\n )
\n
\n return {
\n \"evaluator_approved\": review.approved,
\n \"feedback\": review.feedback,
\n }
\n
\n
\ndef revision_node(state: ResearchState) -> ResearchState:
\n response = model.invoke(
\n f\"\"\"
\n Revise the following technical article.
\n
\n Draft:
\n {state['draft']}
\n
\n Reviewer feedback:
\n {state['feedback']}
\n
\n Preserve correct sections and fix only the identified weaknesses.
\n \"\"\"
\n )
\n
\n return {
\n \"draft\": response.content,
\n \"revision_count\": state.get(\"revision_count\", 0) + 1,
\n }
\n
\n
\ndef human_review_node(state: ResearchState) -> ResearchState:
\n decision = interrupt(
\n {
\n \"message\": \"Review this article before finalization.\",
\n \"draft\": state[\"draft\"],
\n \"automated_feedback\": state.get(\"feedback\", \"\"),
\n \"allowed_actions\": [\"approve\", \"reject\"],
\n }
\n )
\n
\n return {
\n \"human_approved\": decision.get(\"action\") == \"approve\",
\n \"feedback\": decision.get(
\n \"feedback\",
\n state.get(\"feedback\", \"\"),
\n ),
\n }
\n
\n
\ndef finalize_node(state: ResearchState) -> ResearchState:
\n return {\"human_approved\": True}<\/code><\/pre>\nDefine Conditional Routes<\/strong><\/h3>\ndef route_after_evaluation(
\n state: ResearchState,
\n) -> Literal[\"revise\", \"human_review\"]:
\n if state.get(\"evaluator_approved\"):
\n return \"human_review\"
\n
\n if state.get(\"revision_count\", 0) >= 2:
\n return \"human_review\"
\n
\n return \"revise\"
\n
\n
\ndef route_after_human_review(
\n state: ResearchState,
\n) -> Literal[\"finalize\", \"revise\"]:
\n if state.get(\"human_approved\"):
\n return \"finalize\"
\n
\n return \"revise\"<\/code><\/pre>\nThe evaluator does not control the graph directly. It updates the state. The routing function reads that state and selects an allowed edge.<\/p>\n
The revision limit prevents an unbounded evaluator-optimizer loop.<\/p>\n
Construct the Graph<\/strong><\/h3>\nbuilder = StateGraph(ResearchState)
\n
\nbuilder.add_node(\"planner\", planner_node)
\nbuilder.add_node(\"researcher\", researcher_node)
\nbuilder.add_node(\"writer\", writer_node)
\nbuilder.add_node(\"evaluator\", evaluator_node)
\nbuilder.add_node(\"revise\", revision_node)
\nbuilder.add_node(\"human_review\", human_review_node)
\nbuilder.add_node(\"finalize\", finalize_node)
\n
\nbuilder.add_edge(START, \"planner\")
\nbuilder.add_edge(\"planner\", \"researcher\")
\nbuilder.add_edge(\"researcher\", \"writer\")
\nbuilder.add_edge(\"writer\", \"evaluator\")
\n
\nbuilder.add_conditional_edges(
\n \"evaluator\",
\n route_after_evaluation,
\n {
\n \"revise\": \"revise\",
\n \"human_review\": \"human_review\",
\n },
\n)
\n
\nbuilder.add_edge(\"revise\", \"evaluator\")
\n
\nbuilder.add_conditional_edges(
\n \"human_review\",
\n route_after_human_review,
\n {
\n \"finalize\": \"finalize\",
\n \"revise\": \"revise\",
\n },
\n)
\n
\nbuilder.add_edge(\"finalize\", END)
\n
\ncheckpointer = InMemorySaver()
\ngraph = builder.compile(checkpointer=checkpointer)<\/code><\/pre>\n\n<\/figure>\n<\/div>\nRun the Graph<\/strong><\/h3>\nconfig = {
\n \"configurable\": {
\n \"thread_id\": \"graph-engineering-article-001\"
\n }
\n}
\n
\nresult = graph.invoke(
\n {
\n \"topic\": \"Graph engineering for reliable AI agents\",
\n \"revision_count\": 0,
\n },
\n config=config,
\n)
\n
\nif \"__interrupt__\" in result:
\n print(\"The workflow is waiting for human approval.\")<\/code><\/pre>\nResume After Approval<\/strong><\/h3>\nfinal_state = graph.invoke(
\n Command(
\n resume={
\n \"action\": \"approve\",
\n \"feedback\": \"Approved for publication.\",
\n }
\n ),
\n config=config,
\n)
\n
\nprint(final_state[\"draft\"])<\/code><\/pre>\n\n<\/figure>\n<\/div>\nThe same thread_id<\/code> is required because it identifies the stored execution state.<\/p>\nInMemorySaver<\/code> is suitable for demonstration, but it loses all checkpoints when the process restarts. Production systems should use durable persistence backed by a database.<\/p>\nProduction Requirements That Diagrams Often Miss<\/strong><\/h2>\nA graph that runs in a notebook is not automatically production-ready.<\/p>\n
Node Contracts<\/strong><\/h3>\nEvery node should define:<\/p>\n
<\/p>\n- Required inputs<\/li>\n
\n
\n
\n
<\/p>\n
- Produced outputs<\/li>\n
\n
\n
\n
<\/p>\n
- Allowed tools<\/li>\n
\n
\n
\n
<\/p>\n
- Timeout<\/li>\n
\n
\n
\n
<\/p>\n
- Retry policy<\/li>\n
\n
\n
\n
<\/p>\n
- Side effects<\/li>\n
\n
\n
\n
<\/p>\n
- Failure categories<\/li>\n
\n
\n
\n
<\/p>\n
- Validation rules<\/li>\n
\n
\n
\n
<\/p>\n
- Ownership<\/li>\n
\n<\/ul>\n
A node that accepts arbitrary state and returns unstructured text becomes difficult to test.<\/p>\n
Idempotency<\/strong><\/h3>\nA retry should not repeat an irreversible action.<\/p>\n
For example, retrying a payment node must not charge the customer twice.<\/p>\n
Use:<\/p>\n
<\/p>\n- Idempotency keys<\/li>\n
\n
\n
\n
<\/p>\n
- Transaction identifiers<\/li>\n
\n
\n
\n
<\/p>\n
- Deduplication checks<\/li>\n
\n
\n
\n
<\/p>\n
- Operation logs<\/li>\n
\n
\n
\n
<\/p>\n
- Database constraints<\/li>\n
\n<\/ul>\n
Error Classification<\/strong><\/h3>\nNot every failure should be retried.<\/p>\n
Temporary network failure -> RetryA generic retry loop can increase cost without fixing the underlying issue.<\/p>\n
Context Isolation<\/strong><\/h3>\nDo not give every node the complete graph state.<\/p>\n
<\/p>\n- A writer may need evidence and an outline.<\/li>\n
\n
\n
\n
<\/p>\n
- A security reviewer may need code and deployment configuration.<\/li>\n
\n
\n
\n
<\/p>\n
- A publication node may need only the approved artifact and destination.<\/li>\n
\n<\/ol>\n
Context isolation reduces token usage, accidental data exposure, and distraction from irrelevant history.<\/p>\n
Observability<\/strong><\/h3>\nTrace at least:<\/p>\n
<\/p>\n- Node start and completion<\/li>\n
\n
\n
\n
<\/p>\n
- Selected route<\/li>\n
\n
\n
\n
<\/p>\n
- State fields changed<\/li>\n
\n
\n
\n
<\/p>\n
- Tool calls<\/li>\n
\n
\n
\n
<\/p>\n
- Model and prompt version<\/li>\n
\n
\n
\n
<\/p>\n
- Token consumption<\/li>\n
\n
\n
\n
<\/p>\n
- Latency<\/li>\n
\n
\n
\n
<\/p>\n
- Retry count<\/li>\n
\n
\n
\n
<\/p>\n
- Validation results<\/li>\n
\n
\n
\n
<\/p>\n
- Human decisions<\/li>\n
\n
\n
\n
<\/p>\n
- Final outcome<\/li>\n
\n<\/ul>\n
Microsoft Agent Framework also separates dynamic agents from explicitly controlled workflows. Its workflow system supports graph-based execution, typed message routing, conditional paths, parallel processing, checkpoints, and human-in-the-loop interactions.<\/p>\n
Framework Options<\/strong><\/h2>\nLangGraph<\/strong><\/h3>\nBest suited for teams that need low-level control over state, routes, persistence, subgraphs, interrupts, and mixed deterministic-agentic execution.<\/p>\n
Google ADK<\/strong><\/h3>\nUseful for teams building in the Google ecosystem. It supports multi-agent workflows, template-based sequential, parallel, and loop execution, and newer graph-oriented workflows.<\/p>\n
Microsoft Agent Framework<\/strong><\/h3>\nSuitable for Python, .NET, and Go teams that require typed workflows, graph routing, checkpointing, human interaction, and integration with enterprise systems.<\/p>\n
Plain Python and Existing Workflow Engines<\/strong><\/h3>\nA specialized agent framework may be unnecessary when most of the workflow is deterministic.<\/p>\n
Python functions, queues, databases, schedulers, and state-machine libraries can coordinate a small number of LLM-powered steps effectively.<\/p>\n
Limitations<\/strong><\/h2>\nGraph engineering also introduces:<\/p>\n
<\/p>\n- Additional infrastructure<\/li>\n
\n
\n
\n
<\/p>\n
- More state-management complexity<\/li>\n
\n
\n
\n
<\/p>\n
- Higher testing requirements<\/li>\n
\n
\n
\n
<\/p>\n
- Synchronization challenges<\/li>\n
\n
\n
\n
<\/p>\n
- Increased model cost when many agents are used<\/li>\n
\n
\n
\n
<\/p>\n
- More difficult versioning and migrations<\/li>\n
\n
\n
\n
<\/p>\n
- Potential latency at parallel join points<\/li>\n
\n
\n
\n
<\/p>\n
- Overengineering for simple tasks<\/li>\n
\n<\/ul>\n
A graph is useful when its structure makes the system safer, faster, easier to evaluate, or easier to maintain.<\/p>\n
It is not valuable merely because it contains more boxes.<\/p>\n
Conclusion<\/strong><\/h2>\nGraph engineering is not a replacement for prompt engineering, context engineering, harness engineering, or loop engineering.<\/p>\n
It is the layer that coordinates them.<\/p>\n
<\/p>\n- Prompts control individual model calls.<\/li>\n
\n
\n
\n
<\/p>\n
- Context engineering controls what each model sees.<\/li>\n
\n
\n
\n
<\/p>\n
- Agent loops control how an agent reasons and uses tools.<\/li>\n
\n
\n
\n
<\/p>\n
- Graph engineering controls how multiple agents, loops, functions, validators, tools, and humans work together.<\/li>\n
\n<\/ul>\n
The broader lesson is simple:<\/p>\n
<\/p>\n- Do not begin by asking how many agents the system needs.<\/li>\n
\n
\n
\n
<\/p>\n
- Begin by identifying the work, dependencies, decision boundaries, parallel opportunities, validation requirements, failure paths, and human responsibilities.<\/li>\n
\n
\n
\n
<\/p>\n
- The agents fill the nodes.<\/li>\n
\n
\n
\n
<\/p>\n
- The engineering lives in the edges.<\/li>\n
\n<\/ul>\n
Frequently Asked Questions<\/strong><\/h2>\n\nIs graph engineering the same as LangGraph?<\/strong><\/p>\nNo. LangGraph is one framework for implementing graph-based agent workflows. Graph engineering is the broader architectural practice. Are knowledge graphs required? No. Workflow graphs can operate without a knowledge graph. A knowledge graph is useful when the system needs to represent and traverse relationships between entities.<\/p>\n<\/div>\n
Can a graph contain loops?<\/strong><\/p>\nYes. Revision cycles, tool-calling agents, retries, and evaluator-optimizer patterns are loops inside a larger graph.<\/p>\n<\/div>\n
Does every node need an LLM?<\/strong><\/p>\nNo. Most production graphs should combine deterministic functions, tools, policies, and selected LLM-powered nodes.<\/p>\n<\/div>\n
Can one agent be enough?<\/strong><\/p>\nYes. A single agent is often preferable for low-risk, open-ended tasks with a limited toolset and simple recovery requirements.<\/p>\n<\/div>\n<\/div>\n
A node is a bounded unit of execution.<\/p>\n
A node may contain:<\/p>\n
-
<\/p>\n
- An LLM call<\/li>\n
\n
\n
\n <\/p>\n - A complete tool-using agent<\/li>\n
\n
\n
\n <\/p>\n - A Python function<\/li>\n
\n
\n
\n <\/p>\n - A retrieval operation<\/li>\n
\n
\n
\n <\/p>\n - A database query<\/li>\n
\n
\n
\n <\/p>\n - An API request<\/li>\n
\n
\n
\n <\/p>\n - A policy check<\/li>\n
\n
\n
\n <\/p>\n - A test suite<\/li>\n
\n
\n
\n <\/p>\n - A human approval request<\/li>\n
\n
\n
\n <\/p>\n - A subgraph<\/li>\n
\n<\/ul>\n
Not every node should be an AI agent.<\/p>\n
Known business rules should generally remain deterministic. An LLM is useful where semantic interpretation, generation, planning, or ambiguity is involved.<\/p>\n
For example, calculating whether an invoice exceeds an approval threshold does not require an LLM. Understanding whether an email represents a refund request may require one.<\/p>\n
2. Edges<\/strong><\/h3>\n
Edges define which nodes can execute after another node.<\/p>\n
Common edge types include:<\/p>\n
-
<\/p>\n
- Direct edges<\/li>\n
\n
\n
\n <\/p>\n - Conditional edges<\/li>\n
\n
\n
\n <\/p>\n - Parallel edges<\/li>\n
\n
\n
\n <\/p>\n - Looping edges<\/li>\n
\n
\n
\n <\/p>\n - Error edges<\/li>\n
\n
\n
\n <\/p>\n - Human-controlled edges<\/li>\n
\n
\n
\n <\/p>\n - Event-triggered edges<\/li>\n
\n<\/ul>\n
An edge represents a dependency or control rule.<\/p>\n
For example:<\/p>\n
\n<\/figure>\n<\/div>\nThe routing condition may be implemented through deterministic Python logic or an LLM classifier.<\/p>\n
3. State<\/strong><\/h3>\n
State is the shared record carried through the graph.<\/p>\n
It may contain:<\/p>\n
class WorkflowState(TypedDict): \n user_request: str \n task_plan: list[str] \n retrieved_evidence: list[str] \n draft: str \n validation_result: dict \n retry_count: int \n approval_status: str<\/code><\/pre>\nTyped state makes the inputs and outputs of nodes visible. It also reduces the need to pass a complete conversation transcript to every agent.<\/p>\n
LangGraph uses stateful graphs to combine deterministic steps with LLM-driven steps. It also provides persistence, streaming, human-in-the-loop controls, and support for long-running execution.<\/p>\n
4. State Reducers<\/strong><\/h3>\n
Parallel nodes may update the same state field.<\/p>\n
Suppose three research agents return evidence simultaneously:<\/p>\n
{ \n \"evidence\": [\"Source A\"] \n} \n{ \n \"evidence\": [\"Source B\"] \n} \n{ \n \"evidence\": [\"Source C\"] \n}<\/code><\/pre>\nThe graph needs a rule for combining these updates.<\/p>\n
A reducer may append lists, merge dictionaries, select the latest value, or apply a custom conflict-resolution policy.<\/p>\n
Without a clear reducer, parallel updates can overwrite one another or create inconsistent state.<\/p>\n
5. Routes and Guard Conditions<\/strong><\/h3>\n
A route determines which edge should run.<\/p>\n
A guard condition checks whether a transition is allowed.<\/p>\n
def route_after_review(state): \n if state[\"grounding_score\"] <\/code><\/pre>\nHard constraints should not be hidden inside prompts. They should be enforced in routing code whenever possible.<\/p>\n
6. Checkpoints<\/strong><\/h3>\n
A checkpoint stores a snapshot of the graph state.<\/p>\n
Checkpoints allow a workflow to:<\/p>\n
-
<\/p>\n
- Resume after interruption<\/li>\n
\n
\n
\n <\/p>\n - Recover from failure<\/li>\n
\n
\n
\n <\/p>\n - Wait for human feedback<\/li>\n
\n
\n
\n <\/p>\n - Inspect previous states<\/li>\n
\n
\n
\n <\/p>\n - Replay a workflow<\/li>\n
\n
\n
\n <\/p>\n - Support long-running tasks<\/li>\n
\n<\/ul>\n
LangGraph separates thread-level checkpoints from long-term stores. Checkpointers preserve graph state for a specific execution thread, while stores keep application data across threads.<\/p>\n
7. Interrupts<\/strong><\/h3>\n
An interrupt pauses the graph and requests external input.<\/p>\n
Common uses include:<\/p>\n
-
<\/p>\n
- Approval before sending an email<\/li>\n
\n
\n
\n <\/p>\n - Review before publishing content<\/li>\n
\n
\n
\n <\/p>\n - Confirmation before issuing a refund<\/li>\n
\n
\n
\n <\/p>\n - Editing generated parameters<\/li>\n
\n
\n
\n <\/p>\n - Providing missing information<\/li>\n
\n<\/ul>\n
LangGraph interrupts save the current state and allow execution to resume later using the same thread identifier. The documentation also recommends ensuring that side effects before an interrupt are idempotent.<\/p>\n
Important Agent Graph Patterns<\/strong><\/h2>\n
LangGraph and Anthropic describe several recurring orchestration patterns for agentic applications.<\/p>\n
Prompt Chaining<\/strong><\/h3>\n
Each node processes the output of the previous node.<\/p>\n
\u00a0<\/p>\n
\n<\/figure>\n<\/div>\nUse it when the task can be divided into fixed, verifiable stages.<\/p>\n
Routing<\/strong><\/h3>\n
A router sends the request to a specialized branch.<\/p>\n
\u00a0<\/p>\n
\n<\/figure>\n<\/div>\nUse deterministic routing when categories are exact. Use model-based routing when classification requires semantic interpretation.<\/p>\n
Parallelization<\/strong><\/h3>\n
Independent tasks execute concurrently.<\/p>\n
\n<\/figure>\n<\/div>\nParallelization can reduce latency and allow different agents to examine separate dimensions of a problem.<\/p>\n
However, only independent tasks should run in parallel. Creating parallel branches that secretly depend on one another produces incomplete or inconsistent results.<\/p>\n
Orchestrator-Worker<\/strong><\/h3>\n
An orchestrator decomposes a task and delegates parts to workers.<\/p>\n
\u00a0<\/p>\n
\n<\/figure>\n<\/div>\nThis pattern is useful when the number and type of subtasks cannot be known before the request arrives.<\/p>\n
The orchestrator should primarily plan, assign, and integrate. If it directly performs every tool call, the architecture becomes another monolithic agent.<\/p>\n
Evaluator-Optimizer<\/strong><\/h3>\n
One component generates an artifact while another evaluates it.<\/p>\n
\n<\/figure>\n<\/div>\nThis pattern works best when the evaluation criteria are clear and revision produces measurable improvement.<\/p>\n
Human-in-the-Loop<\/strong><\/h3>\n
A human reviews the workflow before a consequential action.<\/p>\n
\n<\/figure>\n<\/div>\nHuman review should be risk-based. Requiring approval for every harmless step makes the system slow without improving safety.<\/p>\n
Hands-On: Building a Graph-Engineered Research Workflow<\/strong><\/h2>\n
We will build a graph that:<\/p>\n
-
<\/p>\n
- Plans the task<\/li>\n
\n
\n
\n <\/p>\n - Collects evidence<\/li>\n
\n
\n
\n <\/p>\n - Writes a draft<\/li>\n
\n
\n
\n <\/p>\n - Evaluates the draft<\/li>\n
\n
\n
\n <\/p>\n - Revises weak drafts<\/li>\n
\n
\n
\n <\/p>\n - Requests human approval<\/li>\n
\n
\n
\n <\/p>\n - Finalizes the output<\/li>\n
\n<\/ol>\n
Install the Dependencies<\/strong><\/h3>\n
pip install -U langgraph langchain langchain-openai<\/code><\/pre>\nInstall the integration package for your chosen model provider separately.<\/p>\n
Set a supported model in the environment:<\/p>\n
model_name =\"openai:gpt-4.1-mini\"<\/code><\/pre>\nDefine the State and Model<\/strong><\/h3>\n
import os \nfrom typing import Literal, TypedDict \n \nfrom pydantic import BaseModel, Field \nfrom langchain.chat_models import init_chat_model \nfrom langgraph.checkpoint.memory import InMemorySaver \nfrom langgraph.graph import END, START, StateGraph \nfrom langgraph.types import Command, interrupt \n \nfrom google.colab import userdata \nos.environ['OPENAI_API_KEY'] = userdata.get('OPENAI_KEY') \n \nmodel = init_chat_model(model_name) \n \n \nclass ResearchState(TypedDict, total=False): \n topic: str \n plan: str \n evidence: str \n draft: str \n feedback: str \n evaluator_approved: bool \n human_approved: bool \n revision_count: int \n \n \nclass ReviewResult(BaseModel): \n approved: bool = Field( \n description=\"Whether the draft is accurate and well grounded.\" \n ) \n feedback: str = Field( \n description=\"Specific corrections required before approval.\" \n ) \n \n \nreview_model = model.with_structured_output(ReviewResult)<\/code><\/pre>\nThe state is the interface shared by all nodes. Each node reads only the values it needs and returns only the fields it o wns.<\/p>\n
Create the Nodes<\/strong><\/h3>\n
def planner_node(state: ResearchState) -> ResearchState: \n response = model.invoke( \n f\"\"\" \n Create a concise research plan for the following topic: \n \n {state['topic']} \n \n Include the questions that must be answered and the evidence needed. \n \"\"\" \n ) \n \n return { \n \"plan\": response.content, \n \"revision_count\": 0, \n } \n \n \ndef researcher_node(state: ResearchState) -> ResearchState: \n response = model.invoke( \n f\"\"\" \n Produce a grounded research brief for this topic: \n \n {state['topic']} \n \n Follow this plan: \n \n {state['plan']} \n \n Clearly separate verified information, assumptions, and open questions. \n \"\"\" \n ) \n \n return {\"evidence\": response.content} \n \n \ndef writer_node(state: ResearchState) -> ResearchState: \n response = model.invoke( \n f\"\"\" \n Write a professional technical article using only the evidence below. \n \n Topic: \n {state['topic']} \n \n Evidence: \n {state['evidence']} \n \n Include an introduction, architecture explanation, implementation \n considerations, limitations, and conclusion. \n \"\"\" \n ) \n \n return {\"draft\": response.content} \n \n \ndef evaluator_node(state: ResearchState) -> ResearchState: \n review = review_model.invoke( \n f\"\"\" \n Evaluate the draft against the available evidence. \n \n Evidence: \n {state['evidence']} \n \n Draft: \n {state['draft']} \n \n Check technical accuracy, grounding, completeness, and clarity. \n \"\"\" \n ) \n \n return { \n \"evaluator_approved\": review.approved, \n \"feedback\": review.feedback, \n } \n \n \ndef revision_node(state: ResearchState) -> ResearchState: \n response = model.invoke( \n f\"\"\" \n Revise the following technical article. \n \n Draft: \n {state['draft']} \n \n Reviewer feedback: \n {state['feedback']} \n \n Preserve correct sections and fix only the identified weaknesses. \n \"\"\" \n ) \n \n return { \n \"draft\": response.content, \n \"revision_count\": state.get(\"revision_count\", 0) + 1, \n } \n \n \ndef human_review_node(state: ResearchState) -> ResearchState: \n decision = interrupt( \n { \n \"message\": \"Review this article before finalization.\", \n \"draft\": state[\"draft\"], \n \"automated_feedback\": state.get(\"feedback\", \"\"), \n \"allowed_actions\": [\"approve\", \"reject\"], \n } \n ) \n \n return { \n \"human_approved\": decision.get(\"action\") == \"approve\", \n \"feedback\": decision.get( \n \"feedback\", \n state.get(\"feedback\", \"\"), \n ), \n } \n \n \ndef finalize_node(state: ResearchState) -> ResearchState: \n return {\"human_approved\": True}<\/code><\/pre>\nDefine Conditional Routes<\/strong><\/h3>\n
def route_after_evaluation( \n state: ResearchState, \n) -> Literal[\"revise\", \"human_review\"]: \n if state.get(\"evaluator_approved\"): \n return \"human_review\" \n \n if state.get(\"revision_count\", 0) >= 2: \n return \"human_review\" \n \n return \"revise\" \n \n \ndef route_after_human_review( \n state: ResearchState, \n) -> Literal[\"finalize\", \"revise\"]: \n if state.get(\"human_approved\"): \n return \"finalize\" \n \n return \"revise\"<\/code><\/pre>\nThe evaluator does not control the graph directly. It updates the state. The routing function reads that state and selects an allowed edge.<\/p>\n
The revision limit prevents an unbounded evaluator-optimizer loop.<\/p>\n
Construct the Graph<\/strong><\/h3>\n
builder = StateGraph(ResearchState) \n \nbuilder.add_node(\"planner\", planner_node) \nbuilder.add_node(\"researcher\", researcher_node) \nbuilder.add_node(\"writer\", writer_node) \nbuilder.add_node(\"evaluator\", evaluator_node) \nbuilder.add_node(\"revise\", revision_node) \nbuilder.add_node(\"human_review\", human_review_node) \nbuilder.add_node(\"finalize\", finalize_node) \n \nbuilder.add_edge(START, \"planner\") \nbuilder.add_edge(\"planner\", \"researcher\") \nbuilder.add_edge(\"researcher\", \"writer\") \nbuilder.add_edge(\"writer\", \"evaluator\") \n \nbuilder.add_conditional_edges( \n \"evaluator\", \n route_after_evaluation, \n { \n \"revise\": \"revise\", \n \"human_review\": \"human_review\", \n }, \n) \n \nbuilder.add_edge(\"revise\", \"evaluator\") \n \nbuilder.add_conditional_edges( \n \"human_review\", \n route_after_human_review, \n { \n \"finalize\": \"finalize\", \n \"revise\": \"revise\", \n }, \n) \n \nbuilder.add_edge(\"finalize\", END) \n \ncheckpointer = InMemorySaver() \ngraph = builder.compile(checkpointer=checkpointer)<\/code><\/pre>\n\n<\/figure>\n<\/div>\nRun the Graph<\/strong><\/h3>\n
config = { \n \"configurable\": { \n \"thread_id\": \"graph-engineering-article-001\" \n } \n} \n \nresult = graph.invoke( \n { \n \"topic\": \"Graph engineering for reliable AI agents\", \n \"revision_count\": 0, \n }, \n config=config, \n) \n \nif \"__interrupt__\" in result: \n print(\"The workflow is waiting for human approval.\")<\/code><\/pre>\nResume After Approval<\/strong><\/h3>\n
final_state = graph.invoke( \n Command( \n resume={ \n \"action\": \"approve\", \n \"feedback\": \"Approved for publication.\", \n } \n ), \n config=config, \n) \n \nprint(final_state[\"draft\"])<\/code><\/pre>\n\n<\/figure>\n<\/div>\nThe same
thread_id<\/code> is required because it identifies the stored execution state.<\/p>\nInMemorySaver<\/code> is suitable for demonstration, but it loses all checkpoints when the process restarts. Production systems should use durable persistence backed by a database.<\/p>\nProduction Requirements That Diagrams Often Miss<\/strong><\/h2>\n
A graph that runs in a notebook is not automatically production-ready.<\/p>\n
Node Contracts<\/strong><\/h3>\n
Every node should define:<\/p>\n
-
<\/p>\n
- Required inputs<\/li>\n
\n
\n
\n <\/p>\n - Produced outputs<\/li>\n
\n
\n
\n <\/p>\n - Allowed tools<\/li>\n
\n
\n
\n <\/p>\n - Timeout<\/li>\n
\n
\n
\n <\/p>\n - Retry policy<\/li>\n
\n
\n
\n <\/p>\n - Side effects<\/li>\n
\n
\n
\n <\/p>\n - Failure categories<\/li>\n
\n
\n
\n <\/p>\n - Validation rules<\/li>\n
\n
\n
\n <\/p>\n - Ownership<\/li>\n
\n<\/ul>\n
A node that accepts arbitrary state and returns unstructured text becomes difficult to test.<\/p>\n
Idempotency<\/strong><\/h3>\n
A retry should not repeat an irreversible action.<\/p>\n
For example, retrying a payment node must not charge the customer twice.<\/p>\n
Use:<\/p>\n
-
<\/p>\n
- Idempotency keys<\/li>\n
\n
\n
\n <\/p>\n - Transaction identifiers<\/li>\n
\n
\n
\n <\/p>\n - Deduplication checks<\/li>\n
\n
\n
\n <\/p>\n - Operation logs<\/li>\n
\n
\n
\n <\/p>\n - Database constraints<\/li>\n
\n<\/ul>\n
Error Classification<\/strong><\/h3>\n
Not every failure should be retried.<\/p>\n
Temporary network failure -> Retry
A generic retry loop can increase cost without fixing the underlying issue.<\/p>\n
Context Isolation<\/strong><\/h3>\n
Do not give every node the complete graph state.<\/p>\n
-
<\/p>\n
- A writer may need evidence and an outline.<\/li>\n
\n
\n
\n <\/p>\n - A security reviewer may need code and deployment configuration.<\/li>\n
\n
\n
\n <\/p>\n - A publication node may need only the approved artifact and destination.<\/li>\n
\n<\/ol>\n
Context isolation reduces token usage, accidental data exposure, and distraction from irrelevant history.<\/p>\n
Observability<\/strong><\/h3>\n
Trace at least:<\/p>\n
-
<\/p>\n
- Node start and completion<\/li>\n
\n
\n
\n <\/p>\n - Selected route<\/li>\n
\n
\n
\n <\/p>\n - State fields changed<\/li>\n
\n
\n
\n <\/p>\n - Tool calls<\/li>\n
\n
\n
\n <\/p>\n - Model and prompt version<\/li>\n
\n
\n
\n <\/p>\n - Token consumption<\/li>\n
\n
\n
\n <\/p>\n - Latency<\/li>\n
\n
\n
\n <\/p>\n - Retry count<\/li>\n
\n
\n
\n <\/p>\n - Validation results<\/li>\n
\n
\n
\n <\/p>\n - Human decisions<\/li>\n
\n
\n
\n <\/p>\n - Final outcome<\/li>\n
\n<\/ul>\n
Microsoft Agent Framework also separates dynamic agents from explicitly controlled workflows. Its workflow system supports graph-based execution, typed message routing, conditional paths, parallel processing, checkpoints, and human-in-the-loop interactions.<\/p>\n
Framework Options<\/strong><\/h2>\n
LangGraph<\/strong><\/h3>\n
Best suited for teams that need low-level control over state, routes, persistence, subgraphs, interrupts, and mixed deterministic-agentic execution.<\/p>\n
Google ADK<\/strong><\/h3>\n
Useful for teams building in the Google ecosystem. It supports multi-agent workflows, template-based sequential, parallel, and loop execution, and newer graph-oriented workflows.<\/p>\n
Microsoft Agent Framework<\/strong><\/h3>\n
Suitable for Python, .NET, and Go teams that require typed workflows, graph routing, checkpointing, human interaction, and integration with enterprise systems.<\/p>\n
Plain Python and Existing Workflow Engines<\/strong><\/h3>\n
A specialized agent framework may be unnecessary when most of the workflow is deterministic.<\/p>\n
Python functions, queues, databases, schedulers, and state-machine libraries can coordinate a small number of LLM-powered steps effectively.<\/p>\n
Limitations<\/strong><\/h2>\n
Graph engineering also introduces:<\/p>\n
-
<\/p>\n
- Additional infrastructure<\/li>\n
\n
\n
\n <\/p>\n - More state-management complexity<\/li>\n
\n
\n
\n <\/p>\n - Higher testing requirements<\/li>\n
\n
\n
\n <\/p>\n - Synchronization challenges<\/li>\n
\n
\n
\n <\/p>\n - Increased model cost when many agents are used<\/li>\n
\n
\n
\n <\/p>\n - More difficult versioning and migrations<\/li>\n
\n
\n
\n <\/p>\n - Potential latency at parallel join points<\/li>\n
\n
\n
\n <\/p>\n - Overengineering for simple tasks<\/li>\n
\n<\/ul>\n
A graph is useful when its structure makes the system safer, faster, easier to evaluate, or easier to maintain.<\/p>\n
It is not valuable merely because it contains more boxes.<\/p>\n
Conclusion<\/strong><\/h2>\n
Graph engineering is not a replacement for prompt engineering, context engineering, harness engineering, or loop engineering.<\/p>\n
It is the layer that coordinates them.<\/p>\n
-
<\/p>\n
- Prompts control individual model calls.<\/li>\n
\n
\n
\n <\/p>\n - Context engineering controls what each model sees.<\/li>\n
\n
\n
\n <\/p>\n - Agent loops control how an agent reasons and uses tools.<\/li>\n
\n
\n
\n <\/p>\n - Graph engineering controls how multiple agents, loops, functions, validators, tools, and humans work together.<\/li>\n
\n<\/ul>\n
The broader lesson is simple:<\/p>\n
-
<\/p>\n
- Do not begin by asking how many agents the system needs.<\/li>\n
\n
\n
\n <\/p>\n - Begin by identifying the work, dependencies, decision boundaries, parallel opportunities, validation requirements, failure paths, and human responsibilities.<\/li>\n
\n
\n
\n <\/p>\n - The agents fill the nodes.<\/li>\n
\n
\n
\n <\/p>\n - The engineering lives in the edges.<\/li>\n
\n<\/ul>\n
Frequently Asked Questions<\/strong><\/h2>\n
\nIs graph engineering the same as LangGraph?<\/strong><\/p>\nNo. LangGraph is one framework for implementing graph-based agent workflows. Graph engineering is the broader architectural practice. Are knowledge graphs required? No. Workflow graphs can operate without a knowledge graph. A knowledge graph is useful when the system needs to represent and traverse relationships between entities.<\/p>\n<\/div>\n
Can a graph contain loops?<\/strong><\/p>\nYes. Revision cycles, tool-calling agents, retries, and evaluator-optimizer patterns are loops inside a larger graph.<\/p>\n<\/div>\n
Does every node need an LLM?<\/strong><\/p>\nNo. Most production graphs should combine deterministic functions, tools, policies, and selected LLM-powered nodes.<\/p>\n<\/div>\n
Can one agent be enough?<\/strong><\/p>\nYes. A single agent is often preferable for low-risk, open-ended tasks with a limited toolset and simple recovery requirements.<\/p>\n<\/div>\n<\/div>\n
- Do not begin by asking how many agents the system needs.<\/li>\n
- Prompts control individual model calls.<\/li>\n
- Additional infrastructure<\/li>\n
- Node start and completion<\/li>\n
- A writer may need evidence and an outline.<\/li>\n
- Idempotency keys<\/li>\n
- Required inputs<\/li>\n
- Plans the task<\/li>\n
- Approval before sending an email<\/li>\n
- Resume after interruption<\/li>\n
- Direct edges<\/li>\n
![\"Harsh](\"https:\/\/av-eks-lekhak.s3.amazonaws.com\/media\/lekhak-profile-images\/converted_image_0fBqNLi.webp\")