Быстрый старт¶
-
Установите пакет и интеграцию LangChain для каждого провайдера, который планируете использовать.
pip install langfuse-hotswap langchain-openai langchain-anthropic -
Создайте промпт в 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— имя одной из моделей, которые вы передадите фабрике; остальные ключи — аргументы конструктора класса этой модели. -
Создайте модели и цепочку один раз — при запуске приложения или на уровне модуля — и используйте цепочку повторно. Это обычный
RunnableLangChain.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") -
Вызовите её. На вход подаётся словарь с переменными промпта, на выходе — сообщение модели.
message = chain.invoke({"product": "Acme Mail", "language": "English", "question": "How do I reset my password?"}) print(message.content) -
Смените модель в Langfuse. Создайте новую версию
support-replyс конфигурацией{"provider": "anthropic", "model": "claude-sonnet-5-5", "temperature": 0.2, "max_tokens": 800}и перенесите на неё метку
production. Следующий вызовchain.invoke(...)уйдёт в Anthropic — без перезапуска и без деплоя. Верните метку обратно, и следующий вызов снова пойдёт в OpenAI. -
При необходимости разберите ответ.
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/: короткие
скрипты, по одному на каждую возможность.