From a0cc396c15b942d3148aae31b4f209315e45cc09 Mon Sep 17 00:00:00 2001 From: Nadir Isweesi Date: Fri, 31 Jul 2026 14:09:27 -0700 Subject: [PATCH 1/2] add multi agent model example --- examples/aio/multi_agent.py | 80 +++++++++++++++++++++++++ examples/aio/reference_to_video.py | 93 ++++++++++++++++++++++++++++++ 2 files changed, 173 insertions(+) create mode 100644 examples/aio/multi_agent.py create mode 100644 examples/aio/reference_to_video.py diff --git a/examples/aio/multi_agent.py b/examples/aio/multi_agent.py new file mode 100644 index 0000000..cf2116a --- /dev/null +++ b/examples/aio/multi_agent.py @@ -0,0 +1,80 @@ +import asyncio +from typing import Sequence + +from absl import app, flags + +import xai_sdk +from xai_sdk.chat import user +from xai_sdk.tools import web_search, x_search + +THINKING = flags.DEFINE_bool("thinking", True, "Whether the model display the reasoning tokens consumed or not") + + +async def basic_response(chat: xai_sdk.aio.chat.Chat): + """Sample and print a complete multi-agent response.""" + # Sample a response from a multi-agent model. + response = await chat.sample() + + print(f"Grok: {response.content}") + + # `cost_usd` is per-request; it shows us the response costs in us dollars. + if response.cost_usd is not None: + print(f"Cost: ${response.cost_usd:.4f}") + + +async def response_with_thinking(chat: xai_sdk.aio.chat.Chat): + """Stream a Multi-agent response with reasoning progress.""" + + print("Grok: ", end="", flush=True) + + content_started = False + last_response = None + + async for response, chunk in chat.stream(): + last_response = response + + if response.usage.reasoning_tokens and not content_started: + print(f"\rThinking...({response.usage.reasoning_tokens} tokens)", end="", flush=True) + + if chunk.content: + if not content_started: + print("\rGrok: ", end="", flush=True) + content_started = True + + print(chunk.content, end="", flush=True) + + print() + + if last_response is None: + raise RuntimeError("The model returned no response.") + + # `cost_usd` is per-request; it shows us the response costs in us dollars. + if last_response.cost_usd is not None: + print(f"Cost: ${last_response.cost_usd:.4f}") + + +async def main(argv: Sequence[str]) -> None: + if len(argv) > 1: + raise app.UsageError("Unexpected command line arguments.") + + client = xai_sdk.AsyncClient() + + chat = client.chat.create( + model="grok-4.20-multi-agent", + # you can choose 4 or 16 agent + agent_count=4, + # you can use tools such as web search, x search, or/and code execution + tools=[web_search(), x_search()], + include=["verbose_streaming"], + ) + + chat.append(user("Research the latest breakthroughs in quantum computing and summarize the key findings.")) + + if THINKING.value: + await response_with_thinking(chat) + else: + await basic_response(chat) + + +if __name__ == "__main__": + app.run(lambda argv: asyncio.run(main(argv))) diff --git a/examples/aio/reference_to_video.py b/examples/aio/reference_to_video.py new file mode 100644 index 0000000..09273bf --- /dev/null +++ b/examples/aio/reference_to_video.py @@ -0,0 +1,93 @@ +import asyncio +from datetime import timedelta +from typing import Sequence, cast + +from absl import app, flags + +import xai_sdk +from xai_sdk.video import VideoAspectRatio, VideoResolution + +MODEL = flags.DEFINE_string("model", "grok-imagine-video", "Video generation model to use.") +IMAGE_URL = flags.DEFINE_string( + "image-url", + "", + "Optional input image (URL or base64 data URL) to use as the first frame (image-to-video).", +) +VIDEO_URL = flags.DEFINE_string( + "video-url", + "", + "Optional input video (URL or base64 data URL) to edit based on the prompt (video-to-video).", +) +DURATION = flags.DEFINE_integer("duration", 0, "Optional duration in seconds (1-15). Use 0 to omit.") +ASPECT_RATIO = flags.DEFINE_string( + "aspect-ratio", + "", + 'Optional aspect ratio. One of: "1:1", "16:9", "9:16", "4:3", "3:4", "3:2", "2:3".', +) +RESOLUTION = flags.DEFINE_string("resolution", "", 'Optional resolution. One of: "480p", "720p".') +REFERENCE_IMAGE_URLS = flags.DEFINE_multi_string( + "reference-image-url", + [], + "Optional reference image URLs for reference-to-video (R2V) generation. Can be specified multiple times.", +) +TIMEOUT = flags.DEFINE_integer("timeout", 600, "Timeout in seconds for polling.") +INTERVAL = flags.DEFINE_integer("interval", 1, "Polling interval in seconds.") + + +async def main(argv: Sequence[str]) -> None: + if len(argv) > 1: + raise app.UsageError("Unexpected command line arguments.") + + client = xai_sdk.AsyncClient() + + duration = DURATION.value or None + image_url = IMAGE_URL.value or None + video_url = VIDEO_URL.value or None + aspect_ratio = cast(VideoAspectRatio, ASPECT_RATIO.value) if ASPECT_RATIO.value else None + resolution = cast(VideoResolution, RESOLUTION.value) if RESOLUTION.value else None + reference_image_urls = REFERENCE_IMAGE_URLS.value or None + + previous_video_url: str | None = video_url + first_turn = True + + while True: + prompt = input("Prompt (blank to stop): " if first_turn else "Edit prompt (blank to stop): ") + if not prompt: + return + + try: + response = await client.video.generate( + prompt=prompt, + model=MODEL.value, + image_url=image_url if first_turn else None, + video_url=previous_video_url, + duration=duration, + aspect_ratio=aspect_ratio, + resolution=resolution, + reference_image_urls=reference_image_urls if first_turn else None, + timeout=timedelta(seconds=TIMEOUT.value), + interval=timedelta(seconds=INTERVAL.value), + ) + print(f"Respects moderation: {response.respect_moderation}") + if response.respect_moderation: + print(f"Video URL: {response.url}") + print(f"Duration: {response.duration}s") + else: + print("Video URL not returned due to moderation.") + if response.cost_usd is not None: + print(f"Cost in USD: ${response.cost_usd:.4f}") + + # Chain edits: use the returned URL as the next input video. + if response.respect_moderation: + previous_video_url = response.url + first_turn = False + except RuntimeError as e: + # request expired + print(e) + except ValueError as e: + # video URL missing from response + print(e) + + +if __name__ == "__main__": + app.run(lambda argv: asyncio.run(main(argv))) From 80c5a0af6c67fe94043df2258f43d4684f71aa29 Mon Sep 17 00:00:00 2001 From: Nadir Isweesi Date: Fri, 31 Jul 2026 14:12:28 -0700 Subject: [PATCH 2/2] add multi agent model example --- examples/aio/reference_to_video.py | 93 ------------------------------ 1 file changed, 93 deletions(-) delete mode 100644 examples/aio/reference_to_video.py diff --git a/examples/aio/reference_to_video.py b/examples/aio/reference_to_video.py deleted file mode 100644 index 09273bf..0000000 --- a/examples/aio/reference_to_video.py +++ /dev/null @@ -1,93 +0,0 @@ -import asyncio -from datetime import timedelta -from typing import Sequence, cast - -from absl import app, flags - -import xai_sdk -from xai_sdk.video import VideoAspectRatio, VideoResolution - -MODEL = flags.DEFINE_string("model", "grok-imagine-video", "Video generation model to use.") -IMAGE_URL = flags.DEFINE_string( - "image-url", - "", - "Optional input image (URL or base64 data URL) to use as the first frame (image-to-video).", -) -VIDEO_URL = flags.DEFINE_string( - "video-url", - "", - "Optional input video (URL or base64 data URL) to edit based on the prompt (video-to-video).", -) -DURATION = flags.DEFINE_integer("duration", 0, "Optional duration in seconds (1-15). Use 0 to omit.") -ASPECT_RATIO = flags.DEFINE_string( - "aspect-ratio", - "", - 'Optional aspect ratio. One of: "1:1", "16:9", "9:16", "4:3", "3:4", "3:2", "2:3".', -) -RESOLUTION = flags.DEFINE_string("resolution", "", 'Optional resolution. One of: "480p", "720p".') -REFERENCE_IMAGE_URLS = flags.DEFINE_multi_string( - "reference-image-url", - [], - "Optional reference image URLs for reference-to-video (R2V) generation. Can be specified multiple times.", -) -TIMEOUT = flags.DEFINE_integer("timeout", 600, "Timeout in seconds for polling.") -INTERVAL = flags.DEFINE_integer("interval", 1, "Polling interval in seconds.") - - -async def main(argv: Sequence[str]) -> None: - if len(argv) > 1: - raise app.UsageError("Unexpected command line arguments.") - - client = xai_sdk.AsyncClient() - - duration = DURATION.value or None - image_url = IMAGE_URL.value or None - video_url = VIDEO_URL.value or None - aspect_ratio = cast(VideoAspectRatio, ASPECT_RATIO.value) if ASPECT_RATIO.value else None - resolution = cast(VideoResolution, RESOLUTION.value) if RESOLUTION.value else None - reference_image_urls = REFERENCE_IMAGE_URLS.value or None - - previous_video_url: str | None = video_url - first_turn = True - - while True: - prompt = input("Prompt (blank to stop): " if first_turn else "Edit prompt (blank to stop): ") - if not prompt: - return - - try: - response = await client.video.generate( - prompt=prompt, - model=MODEL.value, - image_url=image_url if first_turn else None, - video_url=previous_video_url, - duration=duration, - aspect_ratio=aspect_ratio, - resolution=resolution, - reference_image_urls=reference_image_urls if first_turn else None, - timeout=timedelta(seconds=TIMEOUT.value), - interval=timedelta(seconds=INTERVAL.value), - ) - print(f"Respects moderation: {response.respect_moderation}") - if response.respect_moderation: - print(f"Video URL: {response.url}") - print(f"Duration: {response.duration}s") - else: - print("Video URL not returned due to moderation.") - if response.cost_usd is not None: - print(f"Cost in USD: ${response.cost_usd:.4f}") - - # Chain edits: use the returned URL as the next input video. - if response.respect_moderation: - previous_video_url = response.url - first_turn = False - except RuntimeError as e: - # request expired - print(e) - except ValueError as e: - # video URL missing from response - print(e) - - -if __name__ == "__main__": - app.run(lambda argv: asyncio.run(main(argv)))