LangServe Documentation

LangServe provides several automated and developer-friendly features for deploying LangChain objects:

• Automatic Schema Inference: Input and output schemas are automatically inferred from your LangChain object and enforced on every API call with rich error messages.
• Standardized Endpoints: Efficient /invoke, /batch, and /stream endpoints supporting high concurrency.
• Streaming Support:
  • /stream_log: Streams all or some intermediate steps from your chain/agent.
  • /stream_events: (Available since v0.0.40) A simplified way to stream without parsing /stream_log output.
• Interactive Playground: A /playground/ page with streaming output and intermediate steps.
• API Documentation: Automatically generated API docs page with JSONSchema and Swagger.
• Tracing: Built-in optional tracing to LangSmith via API key configuration.
• Client SDK: A client SDK that allows you to call a LangServe server as if it were a local Runnable.

For more advanced features (like persistence, memory, and better debugging), it is recommended to refactor complex LCEL (LangChain Expression Language) chains into discrete LangGraph nodes.

Instead of a single long chain, break the logic into separate nodes for long-running or critical steps (e.g., a retriever_node and a generator_node). This allows you to monitor, debug, and manage the state of each step independently within the StateGraph.

@dataclass
class InputState:
    """Input question from the user."""
    question: str
   
@dataclass 
class OutputState:
    """The output from the graph."""
    answer: str

@dataclass 
class SharedState:
    question: str
    docs: List[str]
    response: str
   
async def retriever_node(state: InputState) -> SharedState:
    """Rettrieve documents based on the user's question."""
    documents = await retriever.ainvoke({"context": state.question})
    return {
        "docs": documents
    }

async def generator_node(state: SharedState) -> OutputState:
    """Generate an answer using an LLM based on the retrieved documents and question."""
    context = " -- DOCUMENT -- ".join(state.docs)
    prompt = [
        SystemMessage(
            content=(
                "Answer the user's question based on the list of documents "
                "that were retrieved. Here are the documents: \n\n"
                f"{context}"
            )
        ),
        HumanMessage(content=state.question),
    ]
    ai_message = await llm.ainvoke(prompt)
    return {"answer": ai_message.content}
    
# Define a new graph
builder = StateGraph(
    SharedState, config_schema=Configuration, input=InputState, output=OutputState
)
builder.add_node("retriever", retriever_node)
builder.add_node("generator", generator_node)
builder.add_edge("__start__", "retriever")
builder.add_edge("retriever", "generator")
graph = builder.compile()
graph.name = "RAG Graph"

LangServe provides a variety of reference implementations in its examples directory to help you get started with different patterns. These include:

• LLMs: Minimal examples using OpenAI and Anthropic with support for async, batching, and streaming.
• Retrieval: Simple retriever servers, Conversational Retrievers, and Configurable Runnables (e.g., changing index names at runtime).
• Agents: Implementations of agents with and without conversation history using OpenAI tools.
• Persistence: Using RunnableWithMessageHistory to implement backend-persisted chat keyed by session_id or conversation_id and user_id.
• Advanced Configuration: Using Configurable Runnable for runtime field and alternative configuration.
• Custom API Patterns: Using APIHandler instead of add_routes for more flexible endpoint definitions and FastAPI integration.
• Authentication: Various patterns including global dependencies, path dependencies, per-request config modifiers, and per-user logic (e.g., searching only within user-owned documents).
• UI/Playground Widgets: Examples of custom widgets for the LangServe playground, such as chat interfaces and file upload widgets.

If you have an existing LangServe application using add_routes, the fastest way to migrate to LangGraph Platform is to wrap your existing Runnable inside a LangGraph node.

This approach involves:

1. Defining InputState, OutputState, and SharedState using @dataclass to represent your original input/output schemas and the internal graph state.
2. Creating an asynchronous node function that calls your existing runnable.ainvoke().
3. Building a StateGraph using the defined states, adding the node, and setting the entrypoint.
4. Compiling the graph.

@dataclass
class InputState:
    input: str
    foo: Optional[str] = None

@dataclass
class OutputState:
    output: Any

@dataclass
class SharedState:
    input: str
    foo: Optional[str] = None
    output: Any
    
runnable = ... # Your existing Runnable

async def my_node(state: InputState, config: RunnableConfig) -> OutputState:
    """Each node does work."""
    return await runnable.ainvoke({"input": state.input, "foo": state.foo})

# Define a new graph
builder = StateGraph(
    SharedState, config_schema=Configuration, input=InputState, output=OutputState
)

# Add the node to the graph
builder.add_node("my_node", my_node)

# Set the entrypoint
builder.add_edge("__start__", "my_node")

# Compile the workflow into an executable graph
graph = builder.compile()
graph.name = "New Graph"