Как протестировать точки входа в функциональный API langgraphPython

Программы на Python
Anonymous
Как протестировать точки входа в функциональный API langgraph

Сообщение Anonymous »

В официальной документации показано только, как тестировать Graph API (StateGraph):
https://docs.langchain.com/oss/python/l ... ng-started
Наша команда обсуждает альтернативы написанию таких тестов для Functional API, особенно в отношении внедрения контрольных точек.
1. Использование компоновщика для передачи контрольной точки
Чтобы протестировать точки входа с помощью InMemorySaver, я решил обернуть каждую функцию точки входа в функцию компоновщика следующим образом:

Код: Выделить всё

def wf1_builder(cp):

@entrypoint(checkpointer=cp)
async def wf1(...):
...

return wf1
Я могу протестировать их следующим образом:

Код: Выделить всё

async def test_wf1:
wf1_test = wf1_builder(InMemorySaver())
result = await wf1_test.ainvoke(...)
Я думаю, что этот подход аналогичен примерам тестов, приведенным в официальной документации для Graph API.
2. Замените контрольный указатель, используя копию
Если я определяю точку входа на 1-м уровне, я могу заменить контрольный указатель, например. с копией, хотя мне это не нравится, потому что выглядит как хак.

Код: Выделить всё

global_checkpointer = build_postgres_saver()

Код: Выделить всё

@entrypoint(checkpointer=global_checkpointer)
async def wf1(...):
...

Код: Выделить всё

async def test_wf1:
wf1_test = wf1.copy({
"checkpointer": InMemorySaver()
})
result = await wf1_test.ainvoke(...)
3. Инициализация контрольной точки в зависимости от среды
Определив разные глобальные контрольные точки в зависимости от среды (тест/продукт), определения точек входа и тесты выглядят чище. Хотя мне это все равно не нравится из-за использования глобальных переменных.

Код: Выделить всё

if os.getenv("ENV") == "test":
global_checkpointer = InMemorySaver()
else:
global_checkpointer = build_postgres_saver()

Код: Выделить всё

@entrypoint(checkpointer=global_checkpointer)
async def wf1(...):
...

Код: Выделить всё

async def test_wf1:
result = await wf1.ainvoke(...)
Какой подход вы рекомендуете? Или какой-то другой подход?

Вернуться в «Python»