diff --git a/README.md b/README.md index 92d0f5a9..3af28936 100644 --- a/README.md +++ b/README.md @@ -208,6 +208,9 @@ Connect to external services. | [Home Automation](docs/examples/home_assistant/) | Control smart home devices | Intermediate | | [RAG Voice Agent](docs/examples/rag/) | Vector search with Annoy + embeddings | Advanced | | [Shopify Voice](complex-agents/shopify-voice-shopper/) | Voice shopping with MCP + Shopify | Advanced | +| [LangChain LangGraph](docs/examples/langchain_langgraph/) | LangGraph StateGraph as a LiveKit LLM backend | Beginner | +| [LangChain Agent](docs/examples/langchain_agent/) | LangChain agent with tools via `create_agent` | Beginner | +| [LangChain Deep Agent](docs/examples/langchain_deepagent/) | Deep agent with planning, subagents, and tools | Intermediate | --- diff --git a/docs/examples/langchain_agent/README.md b/docs/examples/langchain_agent/README.md new file mode 100644 index 00000000..37fdc36a --- /dev/null +++ b/docs/examples/langchain_agent/README.md @@ -0,0 +1,250 @@ +--- +title: LangChain Agent +category: integrations +tags: [langchain, openai] +difficulty: beginner +description: Shows how to use a LangChain agent with tools in a LiveKit voice agent. +demonstrates: + - Using create_agent from langchain.agents to build a tool-calling agent + - Defining LangChain tools with the @tool decorator + - Using a custom llm_node to stream only AI responses and filter tool messages +--- + +This example shows how to use a LangChain agent with tools as the LLM backend for a LiveKit voice agent. The agent is created with `create_agent` from `langchain.agents` and given a `get_weather` tool. A custom `llm_node` override shows how you can modify the graph output if needed. + +**Ask the agent for the weather in your city** + +> **Latency note:** The `LLMAdapter` uses LangGraph's streaming mode to minimise time-to-first-token, but care should be taken when porting LangChain workflows that were not originally designed for voice use cases. For more information on handling long-running operations and providing a better user experience, see the [user feedback documentation](https://docs.livekit.io/agents/logic/external-data/#user-feedback). + +## Prerequisites + +- Add a `.env` in this directory with your LiveKit and OpenAI credentials: + ``` + LIVEKIT_URL=your_livekit_url + LIVEKIT_API_KEY=your_api_key + LIVEKIT_API_SECRET=your_api_secret + OPENAI_API_KEY=your_openai_api_key + ``` +- Install dependencies: + ```bash + pip install "livekit-agents[silero]" livekit-plugins-langchain langchain langchain-openai python-dotenv + ``` + +## Load environment and define the AgentServer + +Import the necessary modules and load environment variables. + +```python +from dotenv import load_dotenv +from langchain.agents import create_agent +from langchain.tools import tool +from langchain_core.messages import AIMessageChunk +from langchain_openai import ChatOpenAI +from livekit.agents import ( + Agent, AgentServer, AgentSession, JobContext, JobProcess, cli, inference, llm, +) +from livekit.plugins import langchain, silero + +load_dotenv() + +server = AgentServer() +``` + +## Prewarm VAD for faster connections + +Preload the VAD model once per process to reduce connection latency. + +```python +def prewarm(proc: JobProcess): + proc.userdata["vad"] = silero.VAD.load() + +server.setup_fnc = prewarm +``` + +## Define a LangChain tool + +Use the `@tool` decorator to define a tool that the LangChain agent can call. Replace the stub implementation with a real API call for production use. Note that this is a [LangChain tool](https://docs.langchain.com/oss/python/langchain/tools), not a LiveKit tool. + +```python +@tool +def get_weather(city: str) -> str: + """Get the current weather for a given city. + + Args: + city: The name of the city to get weather for. + """ + # Stub implementation - replace with a real weather API call + return f"The weather in {city} is sunny and 72 degrees Fahrenheit." +``` + +## Create the LangChain agent + +Use `create_agent` to build a tool-calling agent backed by `ChatOpenAI`. The returned compiled graph can be used with the `LLMAdapter`. + +```python +ASSISTANT_INSTRUCTIONS = """You are a helpful voice AI assistant with access to tools. +You can look up the weather for any city when asked. +Your responses are concise, to the point, and without any complex formatting or punctuation including emojis, asterisks, or other symbols.""" + +_agent_graph = create_agent( + model=ChatOpenAI(model="gpt-4.1-mini", temperature=0.7), + tools=[get_weather], +) +``` + +## Define the agent with a custom llm_node + +Override `llm_node` to stream from the LangChain agent graph directly. The `LLMAdapter` streams all message types including `ToolMessage` content, which would cause tool results to be spoken before the final response. This override filters the stream to only yield `AIMessageChunk` instances. + +```python +class LangChainAgent(Agent): + def __init__(self) -> None: + super().__init__(instructions=ASSISTANT_INSTRUCTIONS) + + async def llm_node(self, chat_ctx, tools, model_settings=None): + state = langchain.LLMAdapter(graph=_agent_graph).chat( + chat_ctx=chat_ctx, tools=[] + ) + lc_messages = state._chat_ctx_to_state() + + async for chunk, _metadata in _agent_graph.astream( + lc_messages, stream_mode="messages" + ): + if isinstance(chunk, AIMessageChunk) and chunk.content: + yield llm.ChatChunk( + id=chunk.id or "", + delta=llm.ChoiceDelta( + role="assistant", content=chunk.content + ), + ) +``` + +## Create the RTC session entrypoint + +Create an `AgentSession` with the LangGraph workflow wrapped in `langchain.LLMAdapter`. The adapter automatically converts LiveKit's chat context to LangChain message types. + +```python +@server.rtc_session() +async def entrypoint(ctx: JobContext): + ctx.log_context_fields = {"room": ctx.room.name} + + session = AgentSession( + stt=inference.STT(model="deepgram/nova-3-general"), + llm=langchain.LLMAdapter(graph=_agent_graph), + tts=inference.TTS(model="cartesia/sonic-3", voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc"), + vad=ctx.proc.userdata["vad"], + ) + + await session.start(agent=LangChainAgent(), room=ctx.room) + await ctx.connect() +``` + +## Run the server + +The `cli.run_app()` function starts the agent server and manages the worker lifecycle. + +```python +if __name__ == "__main__": + cli.run_app(server) +``` + +## Run it + +```console +python langchain_agent.py console +``` + +## How it works + +1. `create_agent` builds a LangGraph-based agent with the `get_weather` tool. +2. When the user asks about weather, the LLM calls the tool and then formulates a response using the result. +3. The custom `llm_node` streams from the graph with `stream_mode="messages"`, filtering to only yield `AIMessageChunk` instances. This prevents `ToolMessage` content from being spoken as duplicate output. +4. The `LLMAdapter` on the session handles converting LiveKit's chat context to LangChain messages via `_chat_ctx_to_state()`. + +## Full example + +```python +from dotenv import load_dotenv +from langchain.agents import create_agent +from langchain.tools import tool +from langchain_core.messages import AIMessageChunk +from langchain_openai import ChatOpenAI +from livekit.agents import ( + Agent, AgentServer, AgentSession, JobContext, JobProcess, cli, inference, llm, +) +from livekit.plugins import langchain, silero + +load_dotenv() + +ASSISTANT_INSTRUCTIONS = """You are a helpful voice AI assistant with access to tools. +You can look up the weather for any city when asked. +Your responses are concise, to the point, and without any complex formatting or punctuation including emojis, asterisks, or other symbols.""" + + +@tool +def get_weather(city: str) -> str: + """Get the current weather for a given city. + + Args: + city: The name of the city to get weather for. + """ + # Stub implementation - replace with a real weather API call + return f"The weather in {city} is sunny and 72 degrees Fahrenheit." + + +_agent_graph = create_agent( + model=ChatOpenAI(model="gpt-4.1-mini", temperature=0.7), + tools=[get_weather], +) + + +class LangChainAgent(Agent): + def __init__(self) -> None: + super().__init__(instructions=ASSISTANT_INSTRUCTIONS) + + async def llm_node(self, chat_ctx, tools, model_settings=None): + state = langchain.LLMAdapter(graph=_agent_graph).chat( + chat_ctx=chat_ctx, tools=[] + ) + lc_messages = state._chat_ctx_to_state() + + async for chunk, _metadata in _agent_graph.astream( + lc_messages, stream_mode="messages" + ): + if isinstance(chunk, AIMessageChunk) and chunk.content: + yield llm.ChatChunk( + id=chunk.id or "", + delta=llm.ChoiceDelta( + role="assistant", content=chunk.content + ), + ) + + +server = AgentServer() + + +def prewarm(proc: JobProcess): + proc.userdata["vad"] = silero.VAD.load() + + +server.setup_fnc = prewarm + + +@server.rtc_session() +async def entrypoint(ctx: JobContext): + ctx.log_context_fields = {"room": ctx.room.name} + + session = AgentSession( + stt=inference.STT(model="deepgram/nova-3-general"), + llm=langchain.LLMAdapter(graph=_agent_graph), + tts=inference.TTS(model="cartesia/sonic-3", voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc"), + vad=ctx.proc.userdata["vad"], + ) + + await session.start(agent=LangChainAgent(), room=ctx.room) + await ctx.connect() + + +if __name__ == "__main__": + cli.run_app(server) +``` diff --git a/docs/examples/langchain_agent/langchain_agent.py b/docs/examples/langchain_agent/langchain_agent.py new file mode 100644 index 00000000..7be492e6 --- /dev/null +++ b/docs/examples/langchain_agent/langchain_agent.py @@ -0,0 +1,109 @@ +""" +--- +title: LangChain Agent +category: integrations +tags: [langchain, openai] +difficulty: beginner +description: Shows how to use a LangChain agent with tools in a LiveKit voice agent. +demonstrates: + - Using create_agent from langchain.agents to build a tool-calling agent + - Defining LangChain tools with the @tool decorator + - Using a custom llm_node to stream only AI responses and filter tool messages +--- + +Latency note: The LLMAdapter uses LangGraph's streaming mode to minimise time-to-first-token, but care should be taken when porting LangChain workflows that were not originally designed for voice use cases. +""" +from dotenv import load_dotenv +from langchain.agents import create_agent +from langchain.tools import tool +from langchain_core.messages import AIMessageChunk +from langchain_openai import ChatOpenAI +from livekit.agents import ( + Agent, + AgentServer, + AgentSession, + JobContext, + JobProcess, + cli, + inference, + llm, +) +from livekit.plugins import langchain, silero + +load_dotenv() + +ASSISTANT_INSTRUCTIONS = """You are a helpful voice AI assistant with access to tools. +You can look up the weather for any city when asked. +Your responses are concise, to the point, and without any complex formatting or punctuation including emojis, asterisks, or other symbols.""" + + +@tool +def get_weather(city: str) -> str: + """Get the current weather for a given city. + + Args: + city: The name of the city to get weather for. + """ + # Stub implementation - replace with a real weather API call + return f"The weather in {city} is sunny and 72 degrees Fahrenheit." + + +_agent_graph = create_agent( + model=ChatOpenAI(model="gpt-4.1-mini", temperature=0.7), + tools=[get_weather], +) + + +class LangChainAgent(Agent): + def __init__(self) -> None: + super().__init__(instructions=ASSISTANT_INSTRUCTIONS) + + async def llm_node(self, chat_ctx, tools, model_settings=None): + """Stream from the LangChain agent graph, yielding only AIMessage chunks. + + The LLMAdapter streams all message types including ToolMessages, this code is responsible for filtering out those ToolMessages so they are not passed to the TTS.""" + state = langchain.LLMAdapter(graph=_agent_graph).chat( + chat_ctx=chat_ctx, tools=[] + ) + # Access the internal state conversion, then stream from the graph + lc_messages = state._chat_ctx_to_state() + + async for chunk, _metadata in _agent_graph.astream( + lc_messages, stream_mode="messages" + ): + if isinstance(chunk, AIMessageChunk) and chunk.content: + yield llm.ChatChunk( + id=chunk.id or "", + delta=llm.ChoiceDelta( + role="assistant", content=chunk.content + ), + ) + + +server = AgentServer() + + +def prewarm(proc: JobProcess): + proc.userdata["vad"] = silero.VAD.load() + + +server.setup_fnc = prewarm + + +@server.rtc_session() +async def entrypoint(ctx: JobContext): + ctx.log_context_fields = {"room": ctx.room.name} + + session = AgentSession( + stt=inference.STT(model="deepgram/nova-3-general"), + llm=langchain.LLMAdapter(graph=_agent_graph), + tts=inference.TTS(model="cartesia/sonic-3", voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc"), + vad=ctx.proc.userdata["vad"], + ) + + await session.start(agent=LangChainAgent(), room=ctx.room) + await ctx.connect() + + +if __name__ == "__main__": + cli.run_app(server) diff --git a/docs/examples/langchain_deepagent/README.md b/docs/examples/langchain_deepagent/README.md new file mode 100644 index 00000000..951fe7ab --- /dev/null +++ b/docs/examples/langchain_deepagent/README.md @@ -0,0 +1,314 @@ +--- +title: LangChain Deep Agent +category: integrations +tags: [langchain, openai, deepagents] +difficulty: intermediate +description: Shows how to use a LangChain deep agent with planning and subagents in a LiveKit voice agent. +demonstrates: + - Using create_deep_agent from the deepagents library + - Delegating work to a subagent for weather research + - Using built-in planning (write_todos) + - Using a custom llm_node to stream only AI responses and filter tool messages +--- + +This example shows how to build a trip planning voice assistant using a LangChain deep agent. The agent uses built-in deep agent capabilities: `write_todos` for breaking a trip plan into steps, and a `weather_researcher` subagent for looking up weather. A custom `get_attractions` tool provides tourist information. + +**Ask the agent to plan your trip to Paris** + +This demonstrates how deep agents go beyond simple tool calling by orchestrating multi-step workflows with planning and delegation — all driven by voice. + +> **Latency note:** The `LLMAdapter` uses LangGraph's streaming mode to minimise time-to-first-token, but care should be taken when porting LangChain workflows that were not originally designed for voice use cases. For more information on handling long-running operations and providing a better user experience, see the [user feedback documentation](https://docs.livekit.io/agents/logic/external-data/#user-feedback). + +## Prerequisites + +- Add a `.env` in this directory with your LiveKit and OpenAI credentials: + ``` + LIVEKIT_URL=your_livekit_url + LIVEKIT_API_KEY=your_api_key + LIVEKIT_API_SECRET=your_api_secret + OPENAI_API_KEY=your_openai_api_key + ``` +- Install dependencies: + ```bash + pip install "livekit-agents[silero]" livekit-plugins-langchain deepagents langchain-openai python-dotenv + ``` + +## Load environment and define the AgentServer + +Import the necessary modules and load environment variables. + +```python +from dotenv import load_dotenv +from deepagents import create_deep_agent +from langchain.tools import tool +from langchain_core.messages import AIMessageChunk +from langchain_openai import ChatOpenAI +from livekit.agents import ( + Agent, AgentServer, AgentSession, JobContext, JobProcess, cli, inference, llm, +) +from livekit.plugins import langchain, silero + +load_dotenv() + +server = AgentServer() +``` + +## Prewarm VAD for faster connections + +Preload the VAD model once per process to reduce connection latency. + +```python +def prewarm(proc: JobProcess): + proc.userdata["vad"] = silero.VAD.load() + +server.setup_fnc = prewarm +``` + +## Define custom tools + +Define tools the agent and its subagents can call. The `get_weather` tool is given to the weather subagent, while `get_attractions` is given to the main agent. + +```python +@tool +def get_weather(city: str) -> str: + """Get the current weather for a given city. + + Args: + city: The name of the city to get weather for. + """ + # Stub implementation - replace with a real weather API call + return f"The weather in {city} is sunny and 72 degrees Fahrenheit." + + +@tool +def get_attractions(city: str) -> str: + """Get popular tourist attractions for a given city. + + Args: + city: The name of the city to get attractions for. + """ + # Stub implementation - replace with a real attractions API call + attractions = { + "paris": "Eiffel Tower, Louvre Museum, Notre-Dame Cathedral, Champs-Elysees", + "london": "Big Ben, Tower of London, British Museum, Buckingham Palace", + "tokyo": "Shibuya Crossing, Senso-ji Temple, Tokyo Tower, Meiji Shrine", + } + result = attractions.get(city.lower(), "Various local landmarks and cultural sites") + return f"Popular attractions in {city}: {result}" +``` + +## Create the deep agent with a subagent + +Use `create_deep_agent` to build the agent. The `subagents` parameter defines a `weather_researcher` that the main agent can delegate to using the built-in `task` tool. The main agent also has access to the built-in `write_todos` tool for planning. + +```python +ASSISTANT_INSTRUCTIONS = """You are a trip planning voice assistant. +When the user asks you to plan a trip, you should: +1. Use write_todos to break the planning into steps. +2. Delegate weather research to the weather_researcher subagent using the task tool. +3. Summarize the plan back to the user. + +Your spoken responses are concise, to the point, and without any complex formatting or punctuation including emojis, asterisks, or other symbols.""" + +_agent_graph = create_deep_agent( + model=ChatOpenAI(model="gpt-4.1-mini", temperature=0.7), + tools=[get_attractions], + system_prompt=ASSISTANT_INSTRUCTIONS, + subagents=[ + { + "name": "weather_researcher", + "description": "Looks up the current weather for a city. Delegate to this subagent when you need weather information for trip planning.", + "system_prompt": "You are a weather research assistant. Use the get_weather tool to look up weather for the requested city and return a brief summary.", + "tools": [get_weather], + }, + ], +) +``` + +## Define the agent with a custom llm_node + +Override `llm_node` to stream from the LangChain agent graph directly. The `LLMAdapter` streams all message types including `ToolMessage` content, which would cause tool results to be spoken before the final response. This override filters the stream to only yield `AIMessageChunk` instances. + +```python +class DeepAgentVoice(Agent): + def __init__(self) -> None: + super().__init__(instructions=ASSISTANT_INSTRUCTIONS) + + async def llm_node(self, chat_ctx, tools, model_settings=None): + state = langchain.LLMAdapter(graph=_agent_graph).chat( + chat_ctx=chat_ctx, tools=[] + ) + lc_messages = state._chat_ctx_to_state() + + async for chunk, _metadata in _agent_graph.astream( + lc_messages, stream_mode="messages" + ): + if isinstance(chunk, AIMessageChunk) and chunk.content: + yield llm.ChatChunk( + id=chunk.id or "", + delta=llm.ChoiceDelta( + role="assistant", content=chunk.content + ), + ) +``` + +## Create the RTC session entrypoint + +Create an `AgentSession` with the LangGraph workflow wrapped in `langchain.LLMAdapter`. The adapter automatically converts LiveKit's chat context to LangChain message types. + +```python +@server.rtc_session() +async def entrypoint(ctx: JobContext): + ctx.log_context_fields = {"room": ctx.room.name} + + session = AgentSession( + stt=inference.STT(model="deepgram/nova-3-general"), + llm=langchain.LLMAdapter(graph=_agent_graph), + tts=inference.TTS(model="cartesia/sonic-3", voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc"), + vad=ctx.proc.userdata["vad"], + ) + + await session.start(agent=DeepAgentVoice(), room=ctx.room) + await ctx.connect() +``` + +## Run the server + +The `cli.run_app()` function starts the agent server and manages the worker lifecycle. + +```python +if __name__ == "__main__": + cli.run_app(server) +``` + +## Run it + +```console +python langchain_deepagent.py console +``` + +Try asking: "Plan a trip to Paris" or "What's the weather in Tokyo?" + +## How it works + +1. `create_deep_agent` builds a LangGraph-based agent with built-in tools for planning and subagent delegation, plus the custom `get_attractions` tool and a `weather_researcher` subagent. +2. When the user asks to plan a trip, the agent uses `write_todos` to break it into steps, delegates weather lookup to the subagent via the `task` tool, and calls `get_attractions` for tourist information. +3. The custom `llm_node` streams from the graph with `stream_mode="messages"`, filtering to only yield `AIMessageChunk` instances. This prevents all the intermediate tool results from being spoken — only the agent's final summary is heard. + +## Full example + +```python +from dotenv import load_dotenv +from deepagents import create_deep_agent +from langchain.tools import tool +from langchain_core.messages import AIMessageChunk +from langchain_openai import ChatOpenAI +from livekit.agents import ( + Agent, AgentServer, AgentSession, JobContext, JobProcess, cli, inference, llm, +) +from livekit.plugins import langchain, silero + +load_dotenv() + +ASSISTANT_INSTRUCTIONS = """You are a trip planning voice assistant. +When the user asks you to plan a trip, you should: +1. Use write_todos to break the planning into steps. +2. Delegate weather research to the weather_researcher subagent using the task tool. +3. Summarize the plan back to the user. + +Your spoken responses are concise, to the point, and without any complex formatting or punctuation including emojis, asterisks, or other symbols.""" + + +@tool +def get_weather(city: str) -> str: + """Get the current weather for a given city. + + Args: + city: The name of the city to get weather for. + """ + # Stub implementation - replace with a real weather API call + return f"The weather in {city} is sunny and 72 degrees Fahrenheit." + + +@tool +def get_attractions(city: str) -> str: + """Get popular tourist attractions for a given city. + + Args: + city: The name of the city to get attractions for. + """ + # Stub implementation - replace with a real attractions API call + attractions = { + "paris": "Eiffel Tower, Louvre Museum, Notre-Dame Cathedral, Champs-Elysees", + "london": "Big Ben, Tower of London, British Museum, Buckingham Palace", + "tokyo": "Shibuya Crossing, Senso-ji Temple, Tokyo Tower, Meiji Shrine", + } + result = attractions.get(city.lower(), "Various local landmarks and cultural sites") + return f"Popular attractions in {city}: {result}" + + +_agent_graph = create_deep_agent( + model=ChatOpenAI(model="gpt-4.1-mini", temperature=0.7), + tools=[get_attractions], + system_prompt=ASSISTANT_INSTRUCTIONS, + subagents=[ + { + "name": "weather_researcher", + "description": "Looks up the current weather for a city. Delegate to this subagent when you need weather information for trip planning.", + "system_prompt": "You are a weather research assistant. Use the get_weather tool to look up weather for the requested city and return a brief summary.", + "tools": [get_weather], + }, + ], +) + + +class DeepAgentVoice(Agent): + def __init__(self) -> None: + super().__init__(instructions=ASSISTANT_INSTRUCTIONS) + + async def llm_node(self, chat_ctx, tools, model_settings=None): + state = langchain.LLMAdapter(graph=_agent_graph).chat( + chat_ctx=chat_ctx, tools=[] + ) + lc_messages = state._chat_ctx_to_state() + + async for chunk, _metadata in _agent_graph.astream( + lc_messages, stream_mode="messages" + ): + if isinstance(chunk, AIMessageChunk) and chunk.content: + yield llm.ChatChunk( + id=chunk.id or "", + delta=llm.ChoiceDelta( + role="assistant", content=chunk.content + ), + ) + + +server = AgentServer() + + +def prewarm(proc: JobProcess): + proc.userdata["vad"] = silero.VAD.load() + + +server.setup_fnc = prewarm + + +@server.rtc_session() +async def entrypoint(ctx: JobContext): + ctx.log_context_fields = {"room": ctx.room.name} + + session = AgentSession( + stt=inference.STT(model="deepgram/nova-3-general"), + llm=langchain.LLMAdapter(graph=_agent_graph), + tts=inference.TTS(model="cartesia/sonic-3", voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc"), + vad=ctx.proc.userdata["vad"], + ) + + await session.start(agent=DeepAgentVoice(), room=ctx.room) + await ctx.connect() + + +if __name__ == "__main__": + cli.run_app(server) +``` diff --git a/docs/examples/langchain_deepagent/langchain_deepagent.py b/docs/examples/langchain_deepagent/langchain_deepagent.py new file mode 100644 index 00000000..e58c52c3 --- /dev/null +++ b/docs/examples/langchain_deepagent/langchain_deepagent.py @@ -0,0 +1,140 @@ +""" +--- +title: LangChain Deep Agent +category: integrations +tags: [langchain, openai, deepagents] +difficulty: intermediate +description: Shows how to use a LangChain deep agent with planning and subagents in a LiveKit voice agent. +demonstrates: + - Using create_deep_agent from the deepagents library + - Delegating work to a subagent for weather research + - Using built-in planning (write_todos) + - Using a custom llm_node to stream only AI responses and filter tool messages +--- + +Latency note: The LLMAdapter uses LangGraph's streaming mode to minimise time-to-first-token, but care should be taken when porting LangChain workflows that were not originally designed for voice use cases. +""" +from dotenv import load_dotenv +from deepagents import create_deep_agent +from langchain.tools import tool +from langchain_core.messages import AIMessageChunk +from langchain_openai import ChatOpenAI +from livekit.agents import ( + Agent, + AgentServer, + AgentSession, + JobContext, + JobProcess, + cli, + inference, + llm, +) +from livekit.plugins import langchain, silero + +load_dotenv() + +ASSISTANT_INSTRUCTIONS = """You are a trip planning voice assistant. +When the user asks you to plan a trip, you should: +1. Use write_todos to break the planning into steps. +2. Delegate weather research to the weather_researcher subagent using the task tool. +3. Summarize the plan back to the user. + +Your spoken responses are concise, to the point, and without any complex formatting or punctuation including emojis, asterisks, or other symbols.""" + + +@tool +def get_weather(city: str) -> str: + """Get the current weather for a given city. + + Args: + city: The name of the city to get weather for. + """ + # Stub implementation - replace with a real weather API call + return f"The weather in {city} is sunny and 72 degrees Fahrenheit." + + +@tool +def get_attractions(city: str) -> str: + """Get popular tourist attractions for a given city. + + Args: + city: The name of the city to get attractions for. + """ + # Stub implementation - replace with a real attractions API call + attractions = { + "paris": "Eiffel Tower, Louvre Museum, Notre-Dame Cathedral, Champs-Elysees", + "london": "Big Ben, Tower of London, British Museum, Buckingham Palace", + "tokyo": "Shibuya Crossing, Senso-ji Temple, Tokyo Tower, Meiji Shrine", + } + result = attractions.get(city.lower(), "Various local landmarks and cultural sites") + return f"Popular attractions in {city}: {result}" + + +_agent_graph = create_deep_agent( + model=ChatOpenAI(model="gpt-4.1-mini", temperature=0.7), + tools=[get_attractions], + system_prompt=ASSISTANT_INSTRUCTIONS, + subagents=[ + { + "name": "weather_researcher", + "description": "Looks up the current weather for a city. Delegate to this subagent when you need weather information for trip planning.", + "system_prompt": "You are a weather research assistant. Use the get_weather tool to look up weather for the requested city and return a brief summary.", + "tools": [get_weather], + }, + ], +) + + +class DeepAgentVoice(Agent): + def __init__(self) -> None: + super().__init__(instructions=ASSISTANT_INSTRUCTIONS) + + async def llm_node(self, chat_ctx, tools, model_settings=None): + """Stream from the deep agent graph, yielding only AIMessage chunks. + + Deep agents use many built-in tools (todos, subagents) and + the llm_node is used to avoid tool results being spoken.""" + state = langchain.LLMAdapter(graph=_agent_graph).chat( + chat_ctx=chat_ctx, tools=[] + ) + lc_messages = state._chat_ctx_to_state() + + async for chunk, _metadata in _agent_graph.astream( + lc_messages, stream_mode="messages" + ): + if isinstance(chunk, AIMessageChunk) and chunk.content: + yield llm.ChatChunk( + id=chunk.id or "", + delta=llm.ChoiceDelta( + role="assistant", content=chunk.content + ), + ) + + +server = AgentServer() + + +def prewarm(proc: JobProcess): + proc.userdata["vad"] = silero.VAD.load() + + +server.setup_fnc = prewarm + + +@server.rtc_session() +async def entrypoint(ctx: JobContext): + ctx.log_context_fields = {"room": ctx.room.name} + + session = AgentSession( + stt=inference.STT(model="deepgram/nova-3-general"), + llm=langchain.LLMAdapter(graph=_agent_graph), + tts=inference.TTS(model="cartesia/sonic-3", voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc"), + vad=ctx.proc.userdata["vad"], + ) + + await session.start(agent=DeepAgentVoice(), room=ctx.room) + await ctx.connect() + + +if __name__ == "__main__": + cli.run_app(server) diff --git a/docs/examples/langchain_langgraph/README.md b/docs/examples/langchain_langgraph/README.md new file mode 100644 index 00000000..d5511056 --- /dev/null +++ b/docs/examples/langchain_langgraph/README.md @@ -0,0 +1,212 @@ +--- +title: LangChain Integration +category: integrations +tags: [langchain, openai] +difficulty: beginner +description: Shows how to use LangGraph to integrate LangChain with LiveKit. +demonstrates: + - Using LangGraph StateGraph to build a conversational workflow + - Using langchain.LLMAdapter to connect a LangGraph workflow to a LiveKit agent + - Using ChatOpenAI from langchain_openai as the LLM provider +--- + +This example shows how to use a LangGraph `StateGraph` as the LLM backend for a LiveKit voice agent via the `livekit-plugins-langchain` plugin. The `LLMAdapter` wraps a compiled LangGraph workflow so it can be used as a drop-in LLM provider in an `AgentSession`. + +> **Latency note:** The `LLMAdapter` uses LangGraph's streaming mode to minimise time-to-first-token, but care should be taken when porting LangChain workflows that were not originally designed for voice use cases. For more information on handling long-running operations and providing a better user experience, see the [user feedback documentation](https://docs.livekit.io/agents/logic/external-data/#user-feedback). + +## Prerequisites + +- Add a `.env` in this directory with your LiveKit and OpenAI credentials: + ``` + LIVEKIT_URL=your_livekit_url + LIVEKIT_API_KEY=your_api_key + LIVEKIT_API_SECRET=your_api_secret + OPENAI_API_KEY=your_openai_api_key + ``` +- Install dependencies: + ```bash + pip install "livekit-agents[silero]" livekit-plugins-langchain langchain-openai langgraph python-dotenv + ``` + +## Load environment and define the AgentServer + +Import the necessary modules, load environment variables, and create an AgentServer. + +```python +from typing import Annotated + +from dotenv import load_dotenv +from langchain_core.messages import SystemMessage +from langchain_openai import ChatOpenAI +from langgraph.graph import END, START, StateGraph, add_messages +from livekit.agents import ( + Agent, AgentServer, AgentSession, JobContext, JobProcess, cli, inference, +) +from livekit.plugins import langchain, silero +from typing_extensions import TypedDict + +load_dotenv() + +server = AgentServer() +``` + +## Prewarm VAD for faster connections + +Preload the VAD model once per process to reduce connection latency. + +```python +def prewarm(proc: JobProcess): + proc.userdata["vad"] = silero.VAD.load() + +server.setup_fnc = prewarm +``` + +## Define the LangGraph workflow + +Build a minimal `StateGraph` with a single node that calls `ChatOpenAI`. The `add_messages` reducer appends new messages to the conversation history, giving the LLM full context on each turn. + +```python +ASSISTANT_INSTRUCTIONS = """You are a helpful voice AI assistant. The user is interacting with you via voice, even if you perceive the conversation as text. +You eagerly assist users with their questions by providing information from your extensive knowledge. +Your responses are concise, to the point, and without any complex formatting or punctuation including emojis, asterisks, or other symbols. +You are curious, friendly, and have a sense of humor.""" + + +class AgentState(TypedDict): + messages: Annotated[list, add_messages] + + +def _create_langgraph_workflow(): + llm = ChatOpenAI(model="gpt-4.1-mini", temperature=0.7) + + def call_model(state: AgentState): + messages = [SystemMessage(content=ASSISTANT_INSTRUCTIONS)] + state["messages"] + response = llm.invoke(messages) + return {"messages": [response]} + + builder = StateGraph(AgentState) + builder.add_node("model", call_model) + builder.add_edge(START, "model") + builder.add_edge("model", END) + + return builder.compile() + + +_langgraph_app = _create_langgraph_workflow() +``` + +## Create the RTC session entrypoint + +Create an `AgentSession` with the LangGraph workflow wrapped in `langchain.LLMAdapter`. The adapter automatically converts LiveKit's chat context to LangChain message types. + +```python +@server.rtc_session() +async def entrypoint(ctx: JobContext): + ctx.log_context_fields = {"room": ctx.room.name} + + session = AgentSession( + stt=inference.STT(model="deepgram/nova-3-general"), + llm=langchain.LLMAdapter(graph=_langgraph_app), + tts=inference.TTS(model="cartesia/sonic-3", voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc"), + vad=ctx.proc.userdata["vad"], + ) + + await session.start(agent=Agent(instructions=ASSISTANT_INSTRUCTIONS), room=ctx.room) + await ctx.connect() +``` + +## Run the server + +The `cli.run_app()` function starts the agent server and manages the worker lifecycle. + +```python +if __name__ == "__main__": + cli.run_app(server) +``` + +## Run it + +```console +python langchain_langraph.py console +``` + +## How it works + +1. A LangGraph `StateGraph` is compiled with a single node that calls `ChatOpenAI`. +2. The compiled graph is wrapped with `langchain.LLMAdapter` so LiveKit can use it as an LLM provider. +3. On each user turn, the adapter converts LiveKit's chat context to LangChain messages and streams the response back. +4. The agent speaks the streamed response via TTS. + +## Full example + +```python +from typing import Annotated + +from dotenv import load_dotenv +from langchain_core.messages import SystemMessage +from langchain_openai import ChatOpenAI +from langgraph.graph import END, START, StateGraph, add_messages +from livekit.agents import ( + Agent, AgentServer, AgentSession, JobContext, JobProcess, cli, inference, +) +from livekit.plugins import langchain, silero +from typing_extensions import TypedDict + +load_dotenv() + +ASSISTANT_INSTRUCTIONS = """You are a helpful voice AI assistant. The user is interacting with you via voice, even if you perceive the conversation as text. +You eagerly assist users with their questions by providing information from your extensive knowledge. +Your responses are concise, to the point, and without any complex formatting or punctuation including emojis, asterisks, or other symbols. +You are curious, friendly, and have a sense of humor.""" + + +class AgentState(TypedDict): + messages: Annotated[list, add_messages] + + +def _create_langgraph_workflow(): + llm = ChatOpenAI(model="gpt-4.1-mini", temperature=0.7) + + def call_model(state: AgentState): + messages = [SystemMessage(content=ASSISTANT_INSTRUCTIONS)] + state["messages"] + response = llm.invoke(messages) + return {"messages": [response]} + + builder = StateGraph(AgentState) + builder.add_node("model", call_model) + builder.add_edge(START, "model") + builder.add_edge("model", END) + + return builder.compile() + + +_langgraph_app = _create_langgraph_workflow() + +server = AgentServer() + + +def prewarm(proc: JobProcess): + proc.userdata["vad"] = silero.VAD.load() + + +server.setup_fnc = prewarm + + +@server.rtc_session() +async def entrypoint(ctx: JobContext): + ctx.log_context_fields = {"room": ctx.room.name} + + session = AgentSession( + stt=inference.STT(model="deepgram/nova-3-general"), + llm=langchain.LLMAdapter(graph=_langgraph_app), + tts=inference.TTS(model="cartesia/sonic-3", voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc"), + vad=ctx.proc.userdata["vad"], + ) + + await session.start(agent=Agent(instructions=ASSISTANT_INSTRUCTIONS), room=ctx.room) + await ctx.connect() + + +if __name__ == "__main__": + cli.run_app(server) +``` diff --git a/docs/examples/langchain_langgraph/langchain_langraph.py b/docs/examples/langchain_langgraph/langchain_langraph.py new file mode 100644 index 00000000..d8588c87 --- /dev/null +++ b/docs/examples/langchain_langgraph/langchain_langraph.py @@ -0,0 +1,93 @@ +""" +--- +title: LangChain Integration +category: integrations +tags: [langchain, openai] +difficulty: beginner +description: Shows how to use LangGraph to integrate LangChain with LiveKit. +demonstrates: + - Using LangGraph StateGraph to build a conversational workflow + - Using langchain.LLMAdapter to connect a LangGraph workflow to a LiveKit agent + - Using ChatOpenAI from langchain_openai as the LLM provider +--- + +Latency note: The LLMAdapter uses LangGraph's streaming mode to minimise time-to-first-token, but care should be taken when porting LangChain workflows that were not originally designed for voice use cases. +""" +from typing import Annotated + +from dotenv import load_dotenv +from langchain_core.messages import SystemMessage +from langchain_openai import ChatOpenAI +from langgraph.graph import END, START, StateGraph, add_messages +from livekit.agents import ( + Agent, + AgentServer, + AgentSession, + JobContext, + JobProcess, + cli, + inference, +) +from livekit.plugins import langchain, silero +from typing_extensions import TypedDict + +load_dotenv() + +ASSISTANT_INSTRUCTIONS = """You are a helpful voice AI assistant. The user is interacting with you via voice, even if you perceive the conversation as text. +You eagerly assist users with their questions by providing information from your extensive knowledge. +Your responses are concise, to the point, and without any complex formatting or punctuation including emojis, asterisks, or other symbols. +You are curious, friendly, and have a sense of humor.""" + + +class AgentState(TypedDict): + """State for the LangGraph agent""" + + messages: Annotated[list, add_messages] + + +def _create_langgraph_workflow(): + """Build a simple LangGraph state graph: messages in -> LLM -> messages out.""" + llm = ChatOpenAI(model="gpt-4.1-mini", temperature=0.7) + + def call_model(state: AgentState): + messages = [SystemMessage(content=ASSISTANT_INSTRUCTIONS)] + state["messages"] + response = llm.invoke(messages) + return {"messages": [response]} + + builder = StateGraph(AgentState) + builder.add_node("model", call_model) + builder.add_edge(START, "model") + builder.add_edge("model", END) + + return builder.compile() + + +_langgraph_app = _create_langgraph_workflow() + +server = AgentServer() + + +def prewarm(proc: JobProcess): + proc.userdata["vad"] = silero.VAD.load() + + +server.setup_fnc = prewarm + + +@server.rtc_session() +async def entrypoint(ctx: JobContext): + ctx.log_context_fields = {"room": ctx.room.name} + + session = AgentSession( + stt=inference.STT(model="deepgram/nova-3-general"), + llm=langchain.LLMAdapter(graph=_langgraph_app), + tts=inference.TTS(model="cartesia/sonic-3", voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc"), + vad=ctx.proc.userdata["vad"], + ) + + await session.start(agent=Agent(instructions=ASSISTANT_INSTRUCTIONS), room=ctx.room) + await ctx.connect() + + +if __name__ == "__main__": + cli.run_app(server) diff --git a/docs/index.yaml b/docs/index.yaml index 4648eaa1..97583365 100644 --- a/docs/index.yaml +++ b/docs/index.yaml @@ -1,6 +1,6 @@ version: '1.0' description: Index of all LiveKit Agent examples with metadata -total_examples: 77 +total_examples: 80 examples: - file_path: complex-agents/avatars/hedra/dynamically_created_avatar/agent.py title: Dynamically Created Avatar @@ -1237,6 +1237,44 @@ examples: - Custom STT configuration with translation capabilities - Event-driven transcription and speech synthesis - Advanced multilingual processing pipeline +- file_path: docs/examples/langchain_langgraph/langchain_langraph.py + title: LangChain Integration + category: integrations + tags: + - langchain + - openai + difficulty: beginner + description: Shows how to use LangGraph to integrate LangChain with LiveKit. + demonstrates: + - Using LangGraph StateGraph to build a conversational workflow + - Using langchain.LLMAdapter to connect a LangGraph workflow to a LiveKit agent + - Using ChatOpenAI from langchain_openai as the LLM provider +- file_path: docs/examples/langchain_agent/langchain_agent.py + title: LangChain Agent + category: integrations + tags: + - langchain + - openai + difficulty: beginner + description: Shows how to use a LangChain agent with tools in a LiveKit voice agent. + demonstrates: + - Using create_agent from langchain.agents to build a tool-calling agent + - Defining LangChain tools with the @tool decorator + - Using a custom llm_node to stream only AI responses and filter tool messages +- file_path: docs/examples/langchain_deepagent/langchain_deepagent.py + title: LangChain Deep Agent + category: integrations + tags: + - langchain + - openai + - deepagents + difficulty: intermediate + description: Shows how to use a LangChain deep agent with planning and subagents in a LiveKit voice agent. + demonstrates: + - Using create_deep_agent from the deepagents library + - Delegating work to a subagent for weather research + - Using built-in planning (write_todos) + - Using a custom llm_node to stream only AI responses and filter tool messages - file_path: moondream_vision/page.mdoc title: Moondream Vision Agent category: vision diff --git a/requirements.txt b/requirements.txt index f6830f25..89ef42db 100644 --- a/requirements.txt +++ b/requirements.txt @@ -11,4 +11,9 @@ websockets>=11.0.3 rich mcp librosa -moondream \ No newline at end of file +moondream +livekit-plugins-langchain +langchain-openai +langgraph +langchain +deepagents \ No newline at end of file