mirror of
https://github.com/hwchase17/langchain.git
synced 2026-04-09 22:12:50 +00:00
973593c5c7
Streaming support is useful if you are doing long-running completions or need interactivity e.g. for chat... adding it to replicate, using a similar pattern to other LLMs that support streaming. Housekeeping: I ran `make format` and `make lint`, no issues reported in the files I touched. I did update the replicate integration test but ran into some issues, specifically: 1. The original test was failing for me due to the model argument not being specified... perhaps this test is not regularly run? I fixed it by adding a call to the lightweight hello world model which should not be burdensome for replicate infra. 2. I couldn't get the `make integration_tests` command to pass... a lot of failures in other integration tests due to missing dependencies... however I did make sure the particluar test file I updated does pass, by running `poetry run pytest tests/integration_tests/llms/test_replicate.py` Finally, I am @tjaffri https://twitter.com/tjaffri for feature announcement tweets... or if you could please tag @docugami https://twitter.com/docugami we would really appreciate that :-) Tagging model maintainers @hwchase17 @baskaryan Thank for all the awesome work you folks are doing. --------- Co-authored-by: Taqi Jaffri <tjaffri@docugami.com>
28 lines
997 B
Python
28 lines
997 B
Python
"""Test Replicate API wrapper."""
|
|
|
|
from langchain.callbacks.manager import CallbackManager
|
|
from langchain.llms.replicate import Replicate
|
|
from tests.unit_tests.callbacks.fake_callback_handler import FakeCallbackHandler
|
|
|
|
TEST_MODEL_NAME = "replicate/hello-world"
|
|
TEST_MODEL_VER = "5c7d5dc6dd8bf75c1acaa8565735e7986bc5b66206b55cca93cb72c9bf15ccaa"
|
|
TEST_MODEL = TEST_MODEL_NAME + ":" + TEST_MODEL_VER
|
|
|
|
|
|
def test_replicate_call() -> None:
|
|
"""Test simple non-streaming call to Replicate."""
|
|
llm = Replicate(model=TEST_MODEL)
|
|
output = llm("LangChain")
|
|
assert output == "hello LangChain"
|
|
|
|
|
|
def test_replicate_streaming_call() -> None:
|
|
"""Test streaming call to Replicate."""
|
|
callback_handler = FakeCallbackHandler()
|
|
callback_manager = CallbackManager([callback_handler])
|
|
|
|
llm = Replicate(streaming=True, callback_manager=callback_manager, model=TEST_MODEL)
|
|
output = llm("LangChain")
|
|
assert output == "hello LangChain"
|
|
assert callback_handler.llm_streams == 15
|