Cookbook: Langserve 集成（SDK v2）

!pip install fastapi sse_starlette httpx langserve "langfuse<3.0.0" langchain-openai langchain

import os
 
# 从项目设置页面获取你的项目密钥
# https://litefuse.cloud
os.environ["LANGFUSE_PUBLIC_KEY"] = ""
os.environ["LANGFUSE_SECRET_KEY"] = ""
os.environ["LANGFUSE_HOST"] = "https://litefuse.cloud"
 
# 你的 openai key
os.environ["OPENAI_API_KEY"] = ""

from langchain_openai import ChatOpenAI
from langchain_core.runnables.config import RunnableConfig
from langfuse import Langfuse
from langfuse.callback import CallbackHandler
from fastapi import FastAPI
from langserve import add_routes
 
langfuse_handler = CallbackHandler()
 
# Tests the SDK connection with the server
langfuse_handler.auth_check()
 
llm = ChatOpenAI()
 
config = RunnableConfig(callbacks=[langfuse_handler])
 
llm_with_langfuse = llm.with_config(config)
 
# Setup server
app = FastAPI()
 
# Add Langserve route
add_routes(
    app,
    llm_with_langfuse,
    path="/test-simple-llm-call",
)

from fastapi.testclient import TestClient
 
# Initialize TestClient
client = TestClient(app)
 
# Test simple route
response = client.post("/test-simple-llm-call/invoke", json={"input": "Tell me a joke?"})

from langchain.prompts import ChatPromptTemplate
from langchain.schema import StrOutputParser
from langserve import add_routes
 
# Create Chain
prompt = ChatPromptTemplate.from_template("Tell me a joke about {topic}")
 
chain = prompt | llm | StrOutputParser()
 
# Add new route
add_routes(
    app,
    chain.with_config(config),
    path="/test-chain",
)
 
# Test chain route
response = client.post("/test-chain/invoke", json={"input": {"topic": "Berlin"}})

from langchain_core.tools import tool
from langchain_core.utils.function_calling import convert_to_openai_tool
from langchain.agents.format_scratchpad.openai_tools import (
    format_to_openai_tool_messages,
)
from langchain.agents import AgentExecutor
from langchain.agents.output_parsers.openai_tools import OpenAIToolsAgentOutputParser
from langserve.pydantic_v1 import BaseModel
from langchain_core.prompts import MessagesPlaceholder
 
class Input(BaseModel):
    input: str
 
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        ("user", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ]
)
 
@tool
def word_length(word: str) -> int:
    """Returns a counter word"""
    return len(word)
 
tools = [word_length]
 
llm_with_tools = llm.bind(tools=[convert_to_openai_tool(tool) for tool in tools])
 
agent = (
    {
        "input": lambda x: x["input"],
        "agent_scratchpad": lambda x: format_to_openai_tool_messages(
            x["intermediate_steps"]
        ),
    }
    | prompt
    | llm_with_tools
    | OpenAIToolsAgentOutputParser()
)
agent_executor = AgentExecutor(agent=agent, tools=tools)
 
agent_config = RunnableConfig({"run_name": "agent"}, callbacks=[langfuse_handler])
 
add_routes(
    app,
    agent_executor.with_types(input_type=Input).with_config(
        agent_config
    ),
    path="/test-agent",
)
 
response = client.post("/test-agent/invoke", json={"input": {"input": "How long is Leonardo DiCaprios last name?"}})