Chapter 73
Constructing a LangGraphAgent
NotebookPython 325 cells
In [1]python · cell 1
python
# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.Constructing a LangGraphAgent
| Author(s) | Pablo Gaeta |
Overview
This notebook demonstrates the necessary code to transform any LangGraph graph into a LangGraphAgent using the concierge.langgraph_server module.
This is a useful interactive way to develop agents before deploying them on a server.
Import dependencies
In [2]python · cell 7
python
import asyncio
import uuid
from typing import TypedDict
from concierge.langgraph_server import langgraph_agent, schemas
from langgraph import graph, typesDefine graph state schema and simple graph implementation
In [3]python · cell 9
python
# State schema
class State(TypedDict, total=False):
id: str
# Node functions
async def set_id(state: State):
new_id = state.get("id")
assert new_id is not None, "must set ID"
await asyncio.sleep(1)
return types.Command(update=State(id=new_id), goto="REVERSE_ID")
async def reverse_id(state: State):
new_id = state.get("id")
assert new_id is not None, "ID must be set before reversing"
await asyncio.sleep(1)
return types.Command(update=State(id=new_id[::-1]))
# Graph
state_graph = graph.StateGraph(state_schema=State)
state_graph.add_node("SET_ID", set_id)
state_graph.add_node("REVERSE_ID", reverse_id)
state_graph.set_entry_point("SET_ID");Build a stateful agent
In [4]python · cell 11
python
basic_agent = langgraph_agent.LangGraphAgent(
state_graph=state_graph,
checkpointer_config=schemas.MemoryBackendConfig(),
)Test: Run a query in debug stream mode
In [5]python · cell 13
python
async for chunk in basic_agent.stream(
input=State(id="hello"),
config={"configurable": {"thread_id": uuid.uuid4().hex}},
stream_mode="debug",
):
print(chunk)Test: Run a query in values mode (full state returned each time)
In [6]python · cell 15
python
async for chunk in basic_agent.stream(
input=State(id="hello"),
config={"configurable": {"thread_id": uuid.uuid4().hex}},
stream_mode="values",
):
print("complete state", "=", chunk)Test: Run a query in updates mode (only state updates returned by each node)
In [7]python · cell 17
python
async for chunk in basic_agent.stream(
input=State(id="hello"),
config={"configurable": {"thread_id": uuid.uuid4().hex}},
stream_mode="updates",
):
print("updates", "->", chunk)Test: Run a query with a list of stream modes (yields stream mode and chunk data)
In [8]python · cell 19
python
async for stream_mode, chunk in basic_agent.stream(
input=State(id="hello"),
config={"configurable": {"thread_id": uuid.uuid4().hex}},
stream_mode=["updates"],
):
print(stream_mode, "->", chunk)Test: Attempt a stateless query, fails since checkpointer requires thread ID. To run stateless queries, must build a stateless agent like the example in the next section.
In [9]python · cell 21
python
async for stream_mode, chunk in basic_agent.stream(
input=State(id="hello"),
config={"configurable": {"thread_id": uuid.uuid4().hex}},
stream_mode=["updates"],
):
print(stream_mode, "->", chunk)Build a stateless agent
In [10]python · cell 23
python
stateless_basic_agent = langgraph_agent.LangGraphAgent(
state_graph=state_graph,
checkpointer_config=None, # no checkpointer means it will only work with stateless requests
)Test: Run stateless run without threads
In [11]python · cell 25
python
async for chunk in stateless_basic_agent.stream(
input=State(id="hello"),
stream_mode="updates",
):
print("complete state", "=", chunk)