Getting Started Quickstart

Quickstart Guide

This guide walks you through migrating an existing LangGraph workflow to AgentMesh in under 2 minutes.

---

Step 1: Define Your TypedDict State

Define your state schema with channel reducers using typing.Annotated:

import operator
from typing import TypedDict, Annotated, List

class WorkflowState(TypedDict):
    query: str
    documents: Annotated[List[str], operator.add]
    answer: str

---

Step 2: Implement Agent Nodes

Nodes are standard Python callables that take the state dictionary and return updates:

def retriever(state: WorkflowState) -> dict:
    q = state["query"]
    return {"documents": [f"Doc 1 for {q}", f"Doc 2 for {q}"]}

def synthesizer(state: WorkflowState) -> dict:
    docs = " ; ".join(state["documents"])
    return {"answer": f"Synthesized answer based on: {docs}"}

---

Step 3: Construct the StateGraph

Use standard LangGraph StateGraph syntax:

from langgraph.graph import StateGraph

builder = StateGraph(WorkflowState)
builder.add_node("retrieve", retriever)
builder.add_node("synthesize", synthesizer)

builder.add_edge("__start__", "retrieve")
builder.add_edge("retrieve", "synthesize")
builder.add_edge("synthesize", "__end__")

---

Step 4: Compile with AgentMesh

Simply replace builder.compile() with agentmesh_adapter.compile(builder):

import agentmesh_adapter

# Compile directly onto bare-metal C++ engine
app = agentmesh_adapter.compile(builder)

---

Step 5: Execute the Workflow

Invoke synchronously or asynchronously:

# Synchronous invocation
output = app.invoke({"query": "Autonomous Agents", "documents": [], "answer": ""})
print("Result:", output["answer"])

# Asynchronous invocation
import asyncio

async def run_async():
    out = await app.ainvoke({"query": "Distributed Consensus", "documents": [], "answer": ""})
    print("Async Result:", out["answer"])

asyncio.run(run_async())
```\n