chore: langchain & pydantic ai weather example

adds a langchain example for weather agent
adds a PydanticAI example for weather agent
changes langchain default event function to send processing events for tool calls instead of message events

Author: Erez Sharim

examples/langchain_simple/agent.py

@@ -1,32 +1,13 @@
-from typing import cast
-
 from flux0_core.agent_runners.api import AgentRunner, Deps, agent_runner
 from flux0_core.agent_runners.context import Context
 from flux0_core.sessions import (
-    MessageEventData,
     StatusEventData,
 )
 from flux0_stream.frameworks.langchain import RunContext, filter_and_map_events, handle_event
 from langchain.chat_models import init_chat_model
 from langchain_core.messages import HumanMessage, SystemMessage
 
-
-async def read_user_input(deps: Deps, context: Context) -> str:
-    """Extract user input from the last message event."""
-    # read session events and expect the last event to be the user input
-    events = await deps.list_session_events(context.session_id)
-    last_event = events[-1]
-    if last_event.type != "message":
-        raise ValueError(f"Expected last event to be a message, got {last_event.type}")
-    user_event_data = cast(MessageEventData, last_event.data)
-    for part in user_event_data["parts"]:
-        if part["type"] == "content":
-            user_input = part["content"]
-            break
-    if not user_input:
-        raise ValueError("No TextPart found in user event data")
-
-    return str(user_input)
+from examples.utils.utils import read_user_input
 
 
 @agent_runner("langchain_simple")

examples/langgraph/__init__.py

@@ -0,0 +1,11 @@
+from lagom import Container
+
+from .weather_agent import WeatherAgentRunner
+
+
+async def init_module(container: Container) -> None:
+    container[WeatherAgentRunner] = WeatherAgentRunner
+
+
+async def shutdown_module() -> None:
+    print("Shutdown!")

examples/langgraph/weather_agent.py

@@ -0,0 +1,85 @@
+from flux0_core.agent_runners.api import AgentRunner, Deps, agent_runner
+from flux0_core.agent_runners.context import Context
+from flux0_core.sessions import (
+    StatusEventData,
+)
+from flux0_stream.frameworks.langchain import RunContext, filter_and_map_events, handle_event
+from langchain_core.runnables.config import RunnableConfig
+from langchain_core.tools import tool
+
+from examples.utils.utils import read_user_input
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.prebuilt import create_react_agent
+
+
+@tool(name_or_callable="get_weather", description="Get the weather for a given city")
+def get_weather(city: str) -> str:
+    """Get weather for a given city."""
+    return f"It's always sunny in {city}!"
+
+
+checkpointer = InMemorySaver()
+weather_agent = create_react_agent(
+    model="gpt-4.1-nano",
+    tools=[get_weather],
+    prompt="You are a friendly, helpful assistant",
+    checkpointer=checkpointer,
+)
+
+
+@agent_runner("langchain_weather_agent")
+class WeatherAgentRunner(AgentRunner):
+    async def run(self, context: Context, deps: Deps) -> bool:
+        # read the agent object
+        agent = await deps.read_agent(context.agent_id)
+        if not agent:
+            raise ValueError(f"Agent with id {context.agent_id} not found")
+
+        # read session events and expect the last event to be the user input
+        user_input = await read_user_input(deps, context)
+        if not user_input:
+            raise ValueError("No user input found in session events")
+
+        deps.logger.info(
+            f"User Input: {user_input} for session {context.session_id} and agent {agent.id}"
+        )
+
+        input = {"messages": [("user", user_input)]}
+        config = RunnableConfig(
+            configurable={
+                "thread_id": context.session_id,
+                "agent_id": agent.id,
+            }
+        )
+        try:
+            model_events = weather_agent.astream_events(
+                input=input,
+                config=config,
+                version="v2",
+            )
+
+            run_ctx: RunContext = RunContext(last_known_event_offset=0)
+            # iterate over the model events and stream them to the client
+            async for e in filter_and_map_events(model_events, deps.logger):
+                await handle_event(
+                    agent,
+                    deps.correlator.correlation_id,
+                    e,
+                    deps.event_emitter,
+                    deps.logger,
+                    run_ctx,
+                )
+
+            return True
+        except Exception as e:
+            await deps.event_emitter.enqueue_status_event(
+                correlation_id=deps.correlator.correlation_id,
+                data=StatusEventData(type="status", status="error", data=str(e)),
+            )
+
+            return False
+        finally:
+            await deps.event_emitter.enqueue_status_event(
+                correlation_id=deps.correlator.correlation_id,
+                data=StatusEventData(type="status", status="completed"),
+            )

examples/pydanticai/__init__.py

@@ -0,0 +1,11 @@
+from lagom import Container
+
+from .weather_agent import WeatherAgentRunner
+
+
+async def init_module(container: Container) -> None:
+    container[WeatherAgentRunner] = WeatherAgentRunner
+
+
+async def shutdown_module() -> None:
+    print("Shutdown!")

examples/pydanticai/weather_agent.py

@@ -0,0 +1,163 @@
+import time
+import uuid
+from dataclasses import dataclass
+from datetime import date
+
+from flux0_core.agent_runners.api import Agent as FAgent
+from flux0_core.agent_runners.api import AgentRunner, Deps, agent_runner
+from flux0_core.agent_runners.context import Context
+from flux0_core.sessions import (
+    EventId,
+    StatusEventData,
+)
+from flux0_stream.emitter.utils.events import send_processing_event
+from flux0_stream.types import ChunkEvent
+from pydantic_ai import Agent
+from pydantic_ai.messages import (
+    FinalResultEvent,
+    FunctionToolCallEvent,
+    FunctionToolResultEvent,
+    PartDeltaEvent,
+    PartStartEvent,
+    TextPartDelta,
+    ToolCallPartDelta,
+)
+from pydantic_ai.tools import RunContext
+
+from examples.utils.utils import read_user_input
+
+
+@dataclass
+class WeatherService:
+    async def get_forecast(self, location: str, forecast_date: date) -> str:
+        # In real code: call weather API, DB queries, etc.
+        return f"The forecast in {location} on {forecast_date} is 24°C and sunny."
+
+    async def get_historic_weather(self, location: str, forecast_date: date) -> str:
+        # In real code: call a historical weather API or DB
+        return f"The weather in {location} on {forecast_date} was 18°C and partly cloudy."
+
+
+weather_agent = Agent[WeatherService, str](
+    "openai:gpt-4.1-nano",
+    deps_type=WeatherService,
+    output_type=str,
+    system_prompt="Providing a weather forecast at the locations the user provides.",
+)
+
+
+@weather_agent.tool
+async def weather_forecast(
+    ctx: RunContext[WeatherService],
+    location: str,
+    forecast_date: date,
+) -> str:
+    if forecast_date >= date.today():
+        return await ctx.deps.get_forecast(location, forecast_date)
+    else:
+        return await ctx.deps.get_historic_weather(location, forecast_date)
+
+
+async def send_message(
+    deps: Deps,
+    input: str,
+    agent: FAgent,
+    event_id: EventId,
+):
+    await send_processing_event(deps, "thinking...")
+    # Begin a node-by-node, streaming iteration
+    async with weather_agent.iter(input, deps=WeatherService()) as run:
+        async for node in run:
+            if Agent.is_model_request_node(node):
+                # A model request node => We can stream tokens from the model's request
+                async with node.stream(run.ctx) as request_stream:
+                    async for event in request_stream:
+                        if isinstance(event, PartStartEvent):
+                            pass
+                        elif isinstance(event, PartDeltaEvent):
+                            if isinstance(event.delta, TextPartDelta):
+                                pass
+                            elif isinstance(event.delta, ToolCallPartDelta):
+                                pass
+                        elif isinstance(event, FinalResultEvent):
+                            pass
+            elif Agent.is_call_tools_node(node):
+                # A handle-response node => The model returned some data, potentially calls a tool
+                async with node.stream(run.ctx) as handle_stream:
+                    async for event in handle_stream:
+                        if isinstance(event, FunctionToolCallEvent):
+                            value = f"[Tools] The LLM calls tool={event.part.tool_name!r} with args={event.part.args}"
+                            await send_processing_event(deps, value)
+                        elif isinstance(event, FunctionToolResultEvent):
+                            value = f"[Tools] Tool call {event.tool_call_id!r} returned => {event.result.content}"
+                            await send_processing_event(deps, value)
+            elif Agent.is_end_node(node):
+                assert run.result.output == node.data.output  # type: ignore
+                # Once an End node is reached, the agent run is complete
+                await deps.event_emitter.enqueue_status_event(
+                    correlation_id=deps.correlator.correlation_id,
+                    data=StatusEventData(type="status", status="typing"),
+                )
+                cec = ChunkEvent(
+                    correlation_id=deps.correlator.correlation_id,
+                    seq=0,
+                    event_id=event_id,
+                    patches=[
+                        {
+                            "op": "add",
+                            "path": "/-",
+                            "value": run.result.output,  # type: ignore
+                        }
+                    ],
+                    metadata={
+                        "agent_id": agent.id,
+                        "agent_name": agent.name,
+                    },
+                    timestamp=time.time(),
+                )
+                await deps.event_emitter.enqueue_event_chunk(cec)
+                await deps.event_emitter.enqueue_status_event(
+                    correlation_id=deps.correlator.correlation_id,
+                    data=StatusEventData(type="status", status="ready"),
+                    event_id=event_id,
+                )
+
+
+@agent_runner("pydantic_weather_agent")
+class WeatherAgentRunner(AgentRunner):
+    async def run(self, context: Context, deps: Deps) -> bool:
+        try:
+            # read the agent object
+            agent = await deps.read_agent(context.agent_id)
+            if not agent:
+                raise ValueError(f"Agent with id {context.agent_id} not found")
+
+            # read session events and expect the last event to be the user input
+            user_input = await read_user_input(deps, context)
+            if not user_input:
+                raise ValueError("No user input found in session events")
+
+            deps.logger.info(
+                f"User Input: {user_input} for session {context.session_id} and agent {agent.id}"
+            )
+
+            await send_message(
+                deps=deps,
+                input=user_input,
+                agent=agent,
+                event_id=EventId(uuid.uuid4().hex),
+            )
+
+            return True
+        except Exception as e:
+            await deps.event_emitter.enqueue_status_event(
+                correlation_id=deps.correlator.correlation_id,
+                data=StatusEventData(type="status", status="error", data=str(e)),
+            )
+
+            return False
+        finally:
+            await deps.event_emitter.enqueue_status_event(
+                correlation_id=deps.correlator.correlation_id,
+                data=StatusEventData(type="status", status="completed"),
+            )

examples/static/static_agent.py

@@ -1,17 +1,18 @@
 import asyncio
 import time
 import uuid
-from typing import cast
 
 from flux0_core.agent_runners.api import AgentRunner, Deps, agent_runner
 from flux0_core.agent_runners.context import Context
 from flux0_core.sessions import (
     EventId,
-    MessageEventData,
     StatusEventData,
 )
+from flux0_stream.emitter.utils.events import send_processing_event
 from flux0_stream.types import ChunkEvent
 
+from examples.utils.utils import read_user_input
+
 
 @agent_runner("static_agent")
 class StaticAgentRunner(AgentRunner):
@@ -21,33 +22,17 @@ async def run(self, context: Context, deps: Deps) -> bool:
         if not agent:
             raise ValueError(f"Agent with id {context.agent_id} not found")
 
-        # read session events and expect the last event to be the user input
-        events = await deps.list_session_events(context.session_id)
-        last_event = events[-1]
-        if last_event.type != "message":
-            return False
-        user_event_data = cast(MessageEventData, last_event.data)
-        for part in user_event_data["parts"]:
-            if part["type"] == "content":
-                user_input = part["content"]
-                break
+        # read session events and extract user input
+        user_input = await read_user_input(deps, context)
         if not user_input:
-            raise ValueError("No TextPart found in user event data")
+            raise ValueError("No user input found in session events")
 
         deps.logger.info(
             f"User Input: {user_input} for session {context.session_id} and agent {agent.id}"
         )
 
         # send processing event, indicating that the agent is thinking
-        await deps.event_emitter.enqueue_status_event(
-            correlation_id=deps.correlator.correlation_id,
-            data=StatusEventData(
-                type="status",
-                status="processing",
-                # optionally include a detailed processing message
-                data={"detail": "Thinking...!"},
-            ),
-        )
+        await send_processing_event(deps, "Thinking...!")
         await asyncio.sleep(1.5)
 
         await deps.event_emitter.enqueue_status_event(

examples/utils/utils.py

@@ -0,0 +1,26 @@
+from typing import cast
+
+from flux0_core.agent_runners.api import Deps
+from flux0_core.agent_runners.context import Context
+from flux0_core.sessions import MessageEventData
+
+
+async def read_user_input(deps: Deps, context: Context) -> str:
+    """Extract user input from the last message event."""
+
+    # read session events and expect the last event to be the user input
+    user_input = None
+
+    events = await deps.list_session_events(context.session_id)
+    last_event = events[-1]
+    if last_event.type != "message":
+        raise ValueError(f"Expected last event to be a message, got {last_event.type}")
+    user_event_data = cast(MessageEventData, last_event.data)
+    for part in user_event_data["parts"]:
+        if part["type"] == "content":
+            user_input = part["content"]
+            break
+    if not user_input:
+        raise ValueError("No TextPart found in user event data")
+
+    return str(user_input)