Перейти к содержанию

Быстрый старт

  1. Установите пакет и интеграцию LangChain для каждого провайдера, который планируете использовать.

    pip install langfuse-hotswap langchain-openai langchain-anthropic
    
  2. Создайте промпт в Langfuse. Чат-промпт с именем support-reply из двух сообщений:

    system: You are a support assistant for {{product}}. Reply in {{language}}.
    user:   {{question}}
    

    и со следующей конфигурацией:

    {"provider": "openai", "model": "gpt-5.5", "temperature": 0.2, "max_tokens": 800}
    

    Присвойте версии метку production. provider — имя одной из моделей, которые вы передадите фабрике; остальные ключи — аргументы конструктора класса этой модели.

  3. Создайте модели и цепочку один раз — при запуске приложения или на уровне модуля — и используйте цепочку повторно. Это обычный Runnable LangChain.

    from langchain_anthropic import ChatAnthropic
    from langchain_openai import ChatOpenAI
    from langfuse import get_client
    
    from langfuse_hotswap import LangfuseChainFactory
    
    factory = LangfuseChainFactory(
        get_client(),
        models={"openai": ChatOpenAI(model="gpt-5.5"), "anthropic": ChatAnthropic(model="claude-sonnet-5-5")},
        allowed_config_keys={"model", "temperature", "max_tokens"},
    )
    chain = factory.create("support-reply")
    
  4. Вызовите её. На вход подаётся словарь с переменными промпта, на выходе — сообщение модели.

    message = chain.invoke({"product": "Acme Mail", "language": "English", "question": "How do I reset my password?"})
    print(message.content)
    
  5. Смените модель в Langfuse. Создайте новую версию support-reply с конфигурацией

    {"provider": "anthropic", "model": "claude-sonnet-5-5", "temperature": 0.2, "max_tokens": 800}
    

    и перенесите на неё метку production. Следующий вызов chain.invoke(...) уйдёт в Anthropic — без перезапуска и без деплоя. Верните метку обратно, и следующий вызов снова пойдёт в OpenAI.

  6. При необходимости разберите ответ. with_structured_output и bind_tools работают так же, как у модели, и возвращают новую цепочку.

    from pydantic import BaseModel
    
    
    class Reply(BaseModel):
        text: str
        escalate_to_human: bool
    
    
    typed_chain = factory.create("support-reply").with_structured_output(Reply)
    reply = typed_chain.invoke({"product": "Acme Mail", "language": "English", "question": "I was charged twice."})
    if reply.escalate_to_human:
        ...
    
Полный пример
# env: LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, LANGFUSE_BASE_URL, OPENAI_API_KEY, ANTHROPIC_API_KEY

from langchain_anthropic import ChatAnthropic
from langchain_openai import ChatOpenAI
from langfuse import get_client
from pydantic import BaseModel

from langfuse_hotswap import LangfuseChainFactory


class Reply(BaseModel):
    text: str
    escalate_to_human: bool


factory = LangfuseChainFactory(
    get_client(),
    models={"openai": ChatOpenAI(model="gpt-5.5"), "anthropic": ChatAnthropic(model="claude-sonnet-5-5")},
    allowed_config_keys={"model", "temperature", "max_tokens"},
)

chain = factory.create("support-reply")
typed_chain = chain.with_structured_output(Reply)

question = {"product": "Acme Mail", "language": "English", "question": "How do I reset my password?"}

print(chain.invoke(question).content)
print(typed_chain.invoke(question).escalate_to_human)

for chunk in chain.stream(question):
    print(chunk.content, end="")

Другие сценарии — в каталоге examples/: короткие скрипты, по одному на каждую возможность.