Getting started with LangGraph and AI Configs

$
pip install launchdarkly-server-sdk-ai launchdarkly-server-sdk-ai-langchain langgraph langchain langchain-core python-dotenv

1
from langchain_core.tools import tool
2
3
4
@tool
5
def get_order_status(order_id: str) -> str:
6
    """Look up the status of a customer order by order ID."""
7
    orders = {
8
        "ORD-123": "Shipped — arrives Thursday",
9
        "ORD-456": "Processing — estimated ship date: tomorrow",
10
        "ORD-789": "Delivered on Monday",
11
    }
12
    return orders.get(order_id, f"No order found with ID {order_id}")

1
import os
2
import ldclient
3
from ldclient import Context
4
from ldclient.config import Config
5
from ldai.client import LDAIClient
6
from ldai.tracker import TokenUsage
7
from ldai_langchain import create_langchain_model, get_ai_metrics_from_response
8
from langchain_core.tools import tool
9
from langgraph.prebuilt import create_react_agent
10
from langgraph.checkpoint.memory import MemorySaver
11
12
13
ldclient.set_config(Config(os.environ.get("LAUNCHDARKLY_SDK_KEY")))
14
15
ai_client = LDAIClient(ldclient.get())
16
17
context = Context.builder("user-123").kind("user").name("Sandy").build()
18
19
# Pass a default for improved resiliency when the AI config is unavailable
20
# or LaunchDarkly is unreachable; omit for a disabled default.
21
# Example:
22
#   from ldai.client import AIAgentConfigDefault
23
#   default = AIAgentConfigDefault(
24
#       enabled=True,
25
#       model={"name": "gpt-5"},
26
#       provider={"name": "openai"},
27
#       instructions="You are a helpful assistant.",
28
#   )
29
#   agent_config = ai_client.agent_config("langgraph-agent", context, default)
30
agent_config = ai_client.agent_config("langgraph-agent", context)
31
32
# create_langchain_model reads agent_config.model.name / .parameters and picks the
33
# right chat model class (OpenAI, Anthropic, …) with no per-provider branching.
34
llm = create_langchain_model(agent_config)
35
36
# MemorySaver gives the ReAct agent short-term memory per thread_id —
37
# follow-up turns on the same thread see the earlier conversation.
38
checkpointer = MemorySaver()
39
agent = create_react_agent(
40
    llm,
41
    [get_order_status],
42
    prompt=agent_config.instructions,
43
    checkpointer=checkpointer,
44
)
45
46
tracker = agent_config.tracker

1
from ldai.tracker import TokenUsage
2
from ldai_langchain import get_ai_metrics_from_response
3
4
5
async def track_langgraph_metrics(tracker, func):
6
    """Track a LangGraph agent run. Aggregates token usage across every message
7
    the agent produced (get_ai_metrics_from_response per message), wrapping
8
    track_duration_of + manual success/tokens/error tracking."""
9
    try:
10
        result = await tracker.track_duration_of(func)
11
        tracker.track_success()
12
        total_in = total_out = total = 0
13
        for message in result.get("messages", []):
14
            metrics = get_ai_metrics_from_response(message)
15
            if metrics.usage:
16
                total_in += metrics.usage.input
17
                total_out += metrics.usage.output
18
                total += metrics.usage.total
19
        if total > 0:
20
            tracker.track_tokens(TokenUsage(input=total_in, output=total_out, total=total))
21
    except Exception:
22
        tracker.track_error()
23
        raise
24
    return result
25
26
27
async def run_turn(agent, tracker, user_input, thread_id):
28
    try:
29
        result = await track_langgraph_metrics(
30
            tracker,
31
            lambda: agent.ainvoke(
32
                {"messages": [{"role": "user", "content": user_input}]},
33
                config={"configurable": {"thread_id": thread_id}},
34
            ),
35
        )
36
        messages = result.get("messages", [])
37
        if messages:
38
            print(f"Agent: {messages[-1].content}")
39
    except Exception as e:
40
        print(f"Error: {e}")
41
42
43
thread_id = "demo-thread"
44
await run_turn(agent, tracker, "What's the status of order ORD-123?", thread_id)
45
await run_turn(agent, tracker, "What about ORD-456?", thread_id)
46
await run_turn(agent, tracker, "Summarize both orders for me.", thread_id)

1
import asyncio
2
import os
3
import ldclient
4
from ldclient import Context
5
from ldclient.config import Config
6
from ldai.client import LDAIClient
7
from ldai.tracker import TokenUsage
8
from ldai_langchain import create_langchain_model, get_ai_metrics_from_response
9
from langchain_core.tools import tool
10
from langgraph.prebuilt import create_react_agent
11
from langgraph.checkpoint.memory import MemorySaver
12
from dotenv import load_dotenv
13
14
load_dotenv()
15
16
SDK_KEY = os.environ.get("LAUNCHDARKLY_SDK_KEY")
17
AGENT_CONFIG_KEY = "langgraph-agent"
18
19
20
@tool
21
def get_order_status(order_id: str) -> str:
22
    """Look up the status of a customer order by order ID."""
23
    orders = {
24
        "ORD-123": "Shipped — arrives Thursday",
25
        "ORD-456": "Processing — estimated ship date: tomorrow",
26
        "ORD-789": "Delivered on Monday",
27
    }
28
    return orders.get(order_id, f"No order found with ID {order_id}")
29
30
31
async def track_langgraph_metrics(tracker, func):
32
    """Track a LangGraph agent run. Aggregates token usage across every message
33
    the agent produced (get_ai_metrics_from_response per message), wrapping
34
    track_duration_of + manual success/tokens/error tracking."""
35
    try:
36
        result = await tracker.track_duration_of(func)
37
        tracker.track_success()
38
        total_in = total_out = total = 0
39
        for message in result.get("messages", []):
40
            metrics = get_ai_metrics_from_response(message)
41
            if metrics.usage:
42
                total_in += metrics.usage.input
43
                total_out += metrics.usage.output
44
                total += metrics.usage.total
45
        if total > 0:
46
            tracker.track_tokens(TokenUsage(input=total_in, output=total_out, total=total))
47
    except Exception:
48
        tracker.track_error()
49
        raise
50
    return result
51
52
53
async def run_turn(agent, tracker, user_input, thread_id):
54
    try:
55
        result = await track_langgraph_metrics(
56
            tracker,
57
            lambda: agent.ainvoke(
58
                {"messages": [{"role": "user", "content": user_input}]},
59
                config={"configurable": {"thread_id": thread_id}},
60
            ),
61
        )
62
        messages = result.get("messages", [])
63
        if messages:
64
            print(f"Agent: {messages[-1].content}")
65
    except Exception as e:
66
        print(f"Error: {e}")
67
68
69
async def async_main():
70
    ldclient.set_config(Config(SDK_KEY))
71
    if not ldclient.get().is_initialized():
72
        print("LaunchDarkly SDK failed to initialize")
73
        return
74
75
    ai_client = LDAIClient(ldclient.get())
76
77
    context = Context.builder("user-123").kind("user").name("Sandy").build()
78
79
    # Pass a default for improved resiliency when the AI config is unavailable
80
    # or LaunchDarkly is unreachable; omit for a disabled default.
81
    # Example:
82
    #   from ldai.client import AIAgentConfigDefault
83
    #   default = AIAgentConfigDefault(
84
    #       enabled=True,
85
    #       model={"name": "gpt-5"},
86
    #       provider={"name": "openai"},
87
    #       instructions="You are a helpful assistant.",
88
    #   )
89
    #   agent_config = ai_client.agent_config(AGENT_CONFIG_KEY, context, default)
90
    agent_config = ai_client.agent_config(AGENT_CONFIG_KEY, context)
91
92
    if not agent_config.enabled:
93
        print("Agent Config is disabled — run the notebook to set up the config")
94
        return
95
96
    # create_langchain_model reads agent_config.model.name / .parameters and picks the
97
    # right chat model class (OpenAI, Anthropic, …) with no per-provider branching.
98
    llm = create_langchain_model(agent_config)
99
100
    # MemorySaver gives the ReAct agent short-term memory per thread_id —
101
    # follow-up turns on the same thread see the earlier conversation.
102
    checkpointer = MemorySaver()
103
    agent = create_react_agent(
104
        llm,
105
        [get_order_status],
106
        prompt=agent_config.instructions,
107
        checkpointer=checkpointer,
108
    )
109
110
    tracker = agent_config.tracker
111
112
    # Three turns on one thread: first fires the tool, second reuses memory
113
    # to answer a follow-up that reuses the tool, third summarizes — no tool.
114
    thread_id = "demo-thread"
115
    await run_turn(agent, tracker, "What's the status of order ORD-123?", thread_id)
116
    await run_turn(agent, tracker, "What about ORD-456?", thread_id)
117
    await run_turn(agent, tracker, "Summarize both orders for me.", thread_id)
118
119
    # Always flush events before closing — trailing events are at risk of being
120
    # lost otherwise, in short-lived scripts and long-running services alike.
121
    ldclient.get().flush()
122
    ldclient.get().close()
123
124
125
def main():
126
    asyncio.run(async_main())
127
128
129
if __name__ == "__main__":
130
    main()