Getting started with Google Gemini

$
pip install launchdarkly-server-sdk-ai
$
pip install google-genai
$
pip install python-dotenv

1
import os
2
from typing import List, Optional, Tuple
3
import ldclient
4
from ldclient import Context
5
from ldclient.config import Config
6
from ldai.client import LDAIClient, LDMessage
7
from ldai.providers.types import LDAIMetrics
8
from ldai.tracker import TokenUsage
9
from google import genai
10
from google.genai import types
11
from dotenv import load_dotenv
12
13
load_dotenv()
14
15
GOOGLE_API_KEY = os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY")
16
SDK_KEY = os.environ.get("LAUNCHDARKLY_SDK_KEY_GEMINI")
17
CONFIG_KEY = "gemini-assistant"
18
AGENT_CONFIG_KEY = "gemini-agent"
19
20
gemini_client = genai.Client(api_key=GOOGLE_API_KEY)
21
ldclient.set_config(Config(SDK_KEY))
22
if not ldclient.get().is_initialized():
23
    exit(1)
24
25
ai_client = LDAIClient(ldclient.get())

1
# Define the context for the current user
2
context = Context.builder("user-123") \
3
    .kind("user") \
4
    .name("Sandy") \
5
    .build()
6
7
# Pass a default for improved resiliency when the AI config is unavailable
8
# or LaunchDarkly is unreachable; omit for a disabled default.
9
# Example:
10
#   from ldai.client import AICompletionConfigDefault
11
#   default = AICompletionConfigDefault(
12
#       enabled=True,
13
#       model={"name": "gemini-2.5-flash"},
14
#       provider={"name": "gemini"},
15
#       messages=[{"role": "system", "content": "You are a helpful assistant."}],
16
#   )
17
#   config = ai_client.completion_config(CONFIG_KEY, context, default, variables={"topic": "Python"})
18
19
# Get the AI Config
20
config = ai_client.completion_config(CONFIG_KEY, context, variables={"topic": "Python"})
21
tracker = config.tracker

1
def gemini_config_kwargs(params):
2
    """Map AI Config parameter names to google.genai GenerateContentConfig arguments.
3
    LaunchDarkly's max_tokens needs to become max_output_tokens; the rest
4
    (temperature, topP, topK, stopSequences) already match the SDK's snake_case aliases.
5
    Drop `tools` — we pass them via the GenerateContentConfig.tools field, so
6
    leaving them in params would duplicate the argument."""
7
    mapping = {"max_tokens": "max_output_tokens"}
8
    return {mapping.get(k, k): v for k, v in (params or {}).items() if k != "tools"}
9
10
def map_to_google_ai_messages(
11
    input_messages: List[LDMessage],
12
) -> Tuple[Optional[str], List[types.Content]]:
13
    """Split LD messages into (system_instruction, history) for google.genai.
14
    System messages are concatenated into a top-level system_instruction;
15
    user/assistant go into Content history (assistant role becomes `model`)."""
16
    history: List[types.Content] = []
17
    system_messages: List[str] = []
18
    for m in input_messages:
19
        if m.role == "system":
20
            system_messages.append(m.content)
21
        elif m.role == "user":
22
            history.append(types.Content(role="user", parts=[types.Part(text=m.content)]))
23
        elif m.role == "assistant":
24
            history.append(types.Content(role="model", parts=[types.Part(text=m.content)]))
25
    system_instruction = " ".join(system_messages) if system_messages else None
26
    return system_instruction, history
27
28
def gemini_metrics(response) -> LDAIMetrics:
29
    """Convert a google.genai response into LDAIMetrics. Passed to
30
    tracker.track_metrics_of, which handles duration + success/error itself."""
31
    usage = getattr(response, "usage_metadata", None)
32
    tokens = None
33
    if usage:
34
        tokens = TokenUsage(
35
            total=usage.total_token_count or 0,
36
            input=usage.prompt_token_count or 0,
37
            output=usage.candidates_token_count or 0,
38
        )
39
    return LDAIMetrics(success=True, usage=tokens)

1
if config.enabled:
2
    tracker = config.tracker
3
    system_instruction, contents = map_to_google_ai_messages(config.messages or [])
4
    contents.append(types.Content(role="user", parts=[types.Part(text="How do I read a file in Python?")]))
5
6
    ld_params = (config.model.to_dict().get("parameters") if config.model else None) or {}
7
8
    response = tracker.track_metrics_of(
9
        lambda: gemini_client.models.generate_content(
10
            model=config.model.name,
11
            contents=contents,
12
            config=types.GenerateContentConfig(
13
                system_instruction=system_instruction,
14
                **gemini_config_kwargs(ld_params),
15
            ),
16
        ),
17
        gemini_metrics,
18
    )

1
# Same fallback pattern as completion — omit the default for a disabled fallback.
2
agent = ai_client.agent_config(AGENT_CONFIG_KEY, context)
3
4
if agent.enabled:
5
    tracker = agent.tracker
6
    ld_params = (agent.model.to_dict().get("parameters") if agent.model else None) or {}
7
8
    # LD returns attached tools under parameters.tools in OpenAI type=function shape.
9
    # Gemini wants {functionDeclarations: [{name, description, parameters}]}.
10
    ld_tools = ld_params.get("tools", []) or []
11
    tools = [{
12
        "functionDeclarations": [
13
            {
14
                "name": t["name"],
15
                "description": t.get("description", ""),
16
                "parameters": t.get("parameters", {"type": "object", "properties": {}}),
17
            }
18
            for t in ld_tools
19
        ]
20
    }] if ld_tools else []
21
22
    # Handlers stay in application code — LD governs the schema, the app owns execution.
23
    def get_order_status(order_id: str) -> str:
24
        orders = {
25
            "ORD-123": "Shipped — arrives Thursday",
26
            "ORD-456": "Processing — estimated ship date: tomorrow",
27
            "ORD-789": "Delivered on Monday",
28
        }
29
        return orders.get(order_id, f"No order found with ID {order_id}")
30
31
    tool_handlers = {"get_order_status": get_order_status}
32
33
    chat = gemini_client.chats.create(
34
        model=agent.model.name,
35
        config=types.GenerateContentConfig(
36
            system_instruction=agent.instructions,
37
            tools=tools,
38
            **gemini_config_kwargs(ld_params),
39
        ),
40
    )
41
42
    # Agent loop: send, handle functionCalls, repeat
43
    MAX_STEPS = 5
44
    message_payload: object = "What's the status of order ORD-123?"
45
    for _ in range(MAX_STEPS):
46
        response = tracker.track_metrics_of(
47
            lambda: chat.send_message(message_payload),
48
            gemini_metrics,
49
        )
50
51
        calls = response.function_calls or []
52
        if not calls:
53
            break
54
55
        function_response_parts = []
56
        for call in calls:
57
            if call.name not in tool_handlers:
58
                raise ValueError(f"Unknown tool: {call.name}")
59
            result = tool_handlers[call.name](**call.args)
60
            tracker.track_tool_call(call.name)
61
            function_response_parts.append(
62
                types.Part(function_response=types.FunctionResponse(name=call.name, response={"result": result}))
63
            )
64
        message_payload = function_response_parts

1
ldclient.get().close()

1
# Always flush events before closing — trailing events are at risk of being
2
# lost otherwise, in short-lived scripts and long-running services alike.
3
ldclient.get().flush()
4
ldclient.get().close()

1
import os
2
from typing import List, Optional, Tuple
3
import ldclient
4
from ldclient import Context
5
from ldclient.config import Config
6
from ldai.client import LDAIClient, LDMessage
7
from ldai.providers.types import LDAIMetrics
8
from ldai.tracker import TokenUsage
9
from google import genai
10
from google.genai import types
11
from dotenv import load_dotenv
12
13
load_dotenv()
14
15
GOOGLE_API_KEY = os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY")
16
SDK_KEY = os.environ.get("LAUNCHDARKLY_SDK_KEY_GEMINI")
17
CONFIG_KEY = "gemini-assistant"
18
AGENT_CONFIG_KEY = "gemini-agent"
19
20
21
def gemini_config_kwargs(params):
22
    """Map AI Config parameter names to google.genai GenerateContentConfig arguments.
23
    LaunchDarkly's max_tokens needs to become max_output_tokens; the rest
24
    (temperature, topP, topK, stopSequences) already match the SDK's snake_case aliases.
25
    Drop `tools` — we pass them via the GenerateContentConfig.tools field, so
26
    leaving them in params would duplicate the argument."""
27
    mapping = {"max_tokens": "max_output_tokens"}
28
    return {mapping.get(k, k): v for k, v in (params or {}).items() if k != "tools"}
29
30
31
def map_to_google_ai_messages(
32
    input_messages: List[LDMessage],
33
) -> Tuple[Optional[str], List[types.Content]]:
34
    """Split LD messages into (system_instruction, history) for google.genai.
35
    System messages are concatenated into a top-level system_instruction;
36
    user/assistant go into Content history (assistant role becomes `model`)."""
37
    history: List[types.Content] = []
38
    system_messages: List[str] = []
39
    for m in input_messages:
40
        if m.role == "system":
41
            system_messages.append(m.content)
42
        elif m.role == "user":
43
            history.append(types.Content(role="user", parts=[types.Part(text=m.content)]))
44
        elif m.role == "assistant":
45
            history.append(types.Content(role="model", parts=[types.Part(text=m.content)]))
46
    system_instruction = " ".join(system_messages) if system_messages else None
47
    return system_instruction, history
48
49
50
def gemini_metrics(response) -> LDAIMetrics:
51
    """Convert a google.genai response into LDAIMetrics. Passed to
52
    tracker.track_metrics_of, which handles duration + success/error itself."""
53
    usage = getattr(response, "usage_metadata", None)
54
    tokens = None
55
    if usage:
56
        tokens = TokenUsage(
57
            total=usage.total_token_count or 0,
58
            input=usage.prompt_token_count or 0,
59
            output=usage.candidates_token_count or 0,
60
        )
61
    return LDAIMetrics(success=True, usage=tokens)
62
63
64
def main():
65
    gemini_client = genai.Client(api_key=GOOGLE_API_KEY)
66
    ldclient.set_config(Config(SDK_KEY))
67
    if not ldclient.get().is_initialized():
68
        return
69
70
    ai_client = LDAIClient(ldclient.get())
71
72
    context = Context.builder("user-123").kind("user").name("Sandy").build()
73
74
    # ===================
75
    # COMPLETION MODE
76
    # ===================
77
78
    # Pass a default for improved resiliency when the AI config is unavailable
79
    # or LaunchDarkly is unreachable; omit for a disabled default.
80
    # Example:
81
    #   from ldai.client import AICompletionConfigDefault
82
    #   default = AICompletionConfigDefault(
83
    #       enabled=True,
84
    #       model={"name": "gemini-2.5-flash"},
85
    #       provider={"name": "gemini"},
86
    #       messages=[{"role": "system", "content": "You are a helpful assistant."}],
87
    #   )
88
    #   config = ai_client.completion_config("gemini-assistant", context, default, variables={"topic": "Python"})
89
    config = ai_client.completion_config(CONFIG_KEY, context, variables={"topic": "Python"})
90
91
    if config.enabled:
92
        tracker = config.tracker
93
        system_instruction, contents = map_to_google_ai_messages(config.messages or [])
94
        contents.append(types.Content(role="user", parts=[types.Part(text="How do I read a file in Python?")]))
95
96
        ld_params = (config.model.to_dict().get("parameters") if config.model else None) or {}
97
98
        tracker.track_metrics_of(
99
            lambda: gemini_client.models.generate_content(
100
                model=config.model.name,
101
                contents=contents,
102
                config=types.GenerateContentConfig(
103
                    system_instruction=system_instruction,
104
                    **gemini_config_kwargs(ld_params),
105
                ),
106
            ),
107
            gemini_metrics,
108
        )
109
110
    # ===================
111
    # AGENT MODE
112
    # ===================
113
114
    # Same fallback pattern as completion — omit the default for a disabled fallback.
115
    agent = ai_client.agent_config(AGENT_CONFIG_KEY, context)
116
117
    if agent.enabled:
118
        tracker = agent.tracker
119
        ld_params = (agent.model.to_dict().get("parameters") if agent.model else None) or {}
120
121
        # LD returns attached tools under parameters.tools in OpenAI type=function shape.
122
        # Gemini wants {functionDeclarations: [{name, description, parameters}]}.
123
        ld_tools = ld_params.get("tools", []) or []
124
        tools = [{
125
            "functionDeclarations": [{
126
                "name": t["name"],
127
                "description": t.get("description", ""),
128
                "parameters": t.get("parameters", {"type": "object", "properties": {}}),
129
            } for t in ld_tools]
130
        }] if ld_tools else []
131
132
        # Handlers stay in application code — LD governs the schema, the app owns execution.
133
        def get_order_status(order_id: str) -> str:
134
            orders = {
135
                "ORD-123": "Shipped — arrives Thursday",
136
                "ORD-456": "Processing — estimated ship date: tomorrow",
137
                "ORD-789": "Delivered on Monday",
138
            }
139
            return orders.get(order_id, f"No order found with ID {order_id}")
140
141
        tool_handlers = {"get_order_status": get_order_status}
142
143
        chat = gemini_client.chats.create(
144
            model=agent.model.name,
145
            config=types.GenerateContentConfig(
146
                system_instruction=agent.instructions,
147
                tools=tools,
148
                **gemini_config_kwargs(ld_params),
149
            ),
150
        )
151
152
        # Agent loop: send, handle functionCalls, repeat
153
        MAX_STEPS = 5
154
        message_payload: object = "What's the status of order ORD-123?"
155
        for _ in range(MAX_STEPS):
156
            response = tracker.track_metrics_of(
157
                lambda: chat.send_message(message_payload),
158
                gemini_metrics,
159
            )
160
161
            calls = response.function_calls or []
162
            if not calls:
163
                break
164
165
            function_response_parts = []
166
            for call in calls:
167
                if call.name not in tool_handlers:
168
                    raise ValueError(f"Unknown tool: {call.name}")
169
                result = tool_handlers[call.name](**call.args)
170
                tracker.track_tool_call(call.name)
171
                function_response_parts.append(
172
                    types.Part(function_response=types.FunctionResponse(name=call.name, response={"result": result}))
173
                )
174
            message_payload = function_response_parts
175
176
    # Always flush events before closing — trailing events are at risk of being
177
    # lost otherwise, in short-lived scripts and long-running services alike.
178
    ldclient.get().flush()
179
    ldclient.get().close()
180
181
182
if __name__ == "__main__":
183
    main()