Also add a new web retrival tab. for that use the webretriver from this documentation:
WebResearchRetriever
class langchain_community.retrievers.web_research.WebResearchRetriever
[source]

Bases: BaseRetriever

Google Search API retriever.

Initialize the retriever.

Note

WebResearchRetriever implements the standard Runnable Interface. 🏃

The Runnable Interface has additional methods that are available on runnables, such as with_types, with_retry, assign, bind, get_graph, and more.

param allow_dangerous_requests: bool = False

    A flag to force users to acknowledge the risks of SSRF attacks when using this retriever.

    Users should set this flag to True if they have taken the necessary precautions to prevent SSRF attacks when using this retriever.

    For example, users can run the requests through a properly configured proxy and prevent the crawler from accidentally crawling internal resources.

param llm_chain: LLMChain [Required]

param metadata: dict[str, Any] | None = None

    Optional metadata associated with the retriever. Defaults to None. This metadata will be associated with each call to this retriever, and passed as arguments to the handlers defined in callbacks. You can use these to eg identify a specific instance of a retriever with its use case.

param num_search_results: int = 1

    Number of pages per Google search

param search: GoogleSearchAPIWrapper [Required]

    Google Search API Wrapper

param tags: list[str] | None = None

    Optional list of tags associated with the retriever. Defaults to None. These tags will be associated with each call to this retriever, and passed as arguments to the handlers defined in callbacks. You can use these to eg identify a specific instance of a retriever with its use case.

param text_splitter: TextSplitter = <langchain_text_splitters.character.RecursiveCharacterTextSplitter object>

    Text splitter for splitting web pages into chunks

param trust_env: bool = False

    Whether to use the http_proxy/https_proxy env variables or check .netrc for proxy configuration

param url_database: List[str] [Optional]

    List of processed URLs

param vectorstore: VectorStore [Required]

    Vector store for storing web pages

async abatch(inputs: list[Input], config: RunnableConfig | list[RunnableConfig] | None = None, *, return_exceptions: bool = False, **kwargs: Any | None) → list[Output]

    Default implementation runs ainvoke in parallel using asyncio.gather.

    The default implementation of batch works well for IO bound runnables.

    Subclasses should override this method if they can batch more efficiently; e.g., if the underlying Runnable uses an API which supports a batch mode.

    Parameters:

            inputs (list[Input]) – A list of inputs to the Runnable.

            config (RunnableConfig | list[RunnableConfig] | None) – A config to use when invoking the Runnable. The config supports standard keys like ‘tags’, ‘metadata’ for tracing purposes, ‘max_concurrency’ for controlling how much work to do in parallel, and other keys. Please refer to the RunnableConfig for more details. Defaults to None.

            return_exceptions (bool) – Whether to return exceptions instead of raising them. Defaults to False.

            kwargs (Any | None) – Additional keyword arguments to pass to the Runnable.

    Returns:

        A list of outputs from the Runnable.
    Return type:

        list[Output]

async abatch_as_completed(inputs: Sequence[Input], config: RunnableConfig | Sequence[RunnableConfig] | None = None, *, return_exceptions: bool = False, **kwargs: Any | None) → AsyncIterator[tuple[int, Output | Exception]]

    Run ainvoke in parallel on a list of inputs, yielding results as they complete.

    Parameters:

            inputs (Sequence[Input]) – A list of inputs to the Runnable.

            config (RunnableConfig | Sequence[RunnableConfig] | None) – A config to use when invoking the Runnable. The config supports standard keys like ‘tags’, ‘metadata’ for tracing purposes, ‘max_concurrency’ for controlling how much work to do in parallel, and other keys. Please refer to the RunnableConfig for more details. Defaults to None. Defaults to None.

            return_exceptions (bool) – Whether to return exceptions instead of raising them. Defaults to False.

            kwargs (Any | None) – Additional keyword arguments to pass to the Runnable.

    Yields:

        A tuple of the index of the input and the output from the Runnable.
    Return type:

        AsyncIterator[tuple[int, Output | Exception]]

async aget_relevant_documents(query: str, *, callbacks: Callbacks = None, tags: list[str] | None = None, metadata: dict[str, Any] | None = None, run_name: str | None = None, **kwargs: Any) → list[Document]

    Deprecated since version langchain-core==0.1.46: Use ainvoke() instead.

    Asynchronously get documents relevant to a query.

    Users should favor using .ainvoke or .abatch rather than aget_relevant_documents directly.

    Parameters:

            query (str) – string to find relevant documents for.

            callbacks (Callbacks) – Callback manager or list of callbacks.

            tags (Optional[list[str]]) – Optional list of tags associated with the retriever. These tags will be associated with each call to this retriever, and passed as arguments to the handlers defined in callbacks. Defaults to None.

            metadata (Optional[dict[str, Any]]) – Optional metadata associated with the retriever. This metadata will be associated with each call to this retriever, and passed as arguments to the handlers defined in callbacks. Defaults to None.

            run_name (Optional[str]) – Optional name for the run. Defaults to None.

            kwargs (Any) – Additional arguments to pass to the retriever.

    Returns:

        List of relevant documents.
    Return type:

        list[Document]

async ainvoke(input: str, config: RunnableConfig | None = None, **kwargs: Any) → list[Document]

Asynchronously invoke the retriever to get relevant documents.

Main entry point for asynchronous retriever invocations.

Parameters:

        input (str) – The query string.

        config (RunnableConfig | None) – Configuration for the retriever. Defaults to None.

        kwargs (Any) – Additional arguments to pass to the retriever.

Returns:

    List of relevant documents.
Return type:

    list[Document]

Examples:

await retriever.ainvoke("query")

async astream(input: Input, config: RunnableConfig | None = None, **kwargs: Any | None) → AsyncIterator[Output]

    Default implementation of astream, which calls ainvoke. Subclasses should override this method if they support streaming output.

    Parameters:

            input (Input) – The input to the Runnable.

            config (RunnableConfig | None) – The config to use for the Runnable. Defaults to None.

            kwargs (Any | None) – Additional keyword arguments to pass to the Runnable.

    Yields:

        The output of the Runnable.
    Return type:

        AsyncIterator[Output]

async astream_events(input: Any, config: RunnableConfig | None = None, *, version: Literal['v1', 'v2'], include_names: Sequence[str] | None = None, include_types: Sequence[str] | None = None, include_tags: Sequence[str] | None = None, exclude_names: Sequence[str] | None = None, exclude_types: Sequence[str] | None = None, exclude_tags: Sequence[str] | None = None, **kwargs: Any) → AsyncIterator[StandardStreamEvent | CustomStreamEvent]

Generate a stream of events.

Use to create an iterator over StreamEvents that provide real-time information about the progress of the Runnable, including StreamEvents from intermediate results.

A StreamEvent is a dictionary with the following schema:

    event: str - Event names are of the

        format: on_[runnable_type]_(start|stream|end).

    name: str - The name of the Runnable that generated the event.

    run_id: str - randomly generated ID associated with the given execution of

        the Runnable that emitted the event. A child Runnable that gets invoked as part of the execution of a parent Runnable is assigned its own unique ID.

    parent_ids: List[str] - The IDs of the parent runnables that

        generated the event. The root Runnable will have an empty list. The order of the parent IDs is from the root to the immediate parent. Only available for v2 version of the API. The v1 version of the API will return an empty list.

    tags: Optional[List[str]] - The tags of the Runnable that generated

        the event.

    metadata: Optional[Dict[str, Any]] - The metadata of the Runnable

        that generated the event.

    data: Dict[str, Any]

Below is a table that illustrates some evens that might be emitted by various chains. Metadata fields have been omitted from the table for brevity. Chain definitions have been included after the table.

ATTENTION This reference table is for the V2 version of the schema.

event
	

name
	

chunk
	

input
	

output

on_chat_model_start
	

[model name]
		

{“messages”: [[SystemMessage, HumanMessage]]}
	

on_chat_model_stream
	

[model name]
	

AIMessageChunk(content=”hello”)
		

on_chat_model_end
	

[model name]
		

{“messages”: [[SystemMessage, HumanMessage]]}
	

AIMessageChunk(content=”hello world”)

on_llm_start
	

[model name]
		

{‘input’: ‘hello’}
	

on_llm_stream
	

[model name]
	

‘Hello’
		

on_llm_end
	

[model name]
		

‘Hello human!’
	

on_chain_start
	

format_docs
			

on_chain_stream
	

format_docs
	

“hello world!, goodbye world!”
		

on_chain_end
	

format_docs
		

[Document(…)]
	

“hello world!, goodbye world!”

on_tool_start
	

some_tool
		

{“x”: 1, “y”: “2”}
	

on_tool_end
	

some_tool
			

{“x”: 1, “y”: “2”}

on_retriever_start
	

[retriever name]
		

{“query”: “hello”}
	

on_retriever_end
	

[retriever name]
		

{“query”: “hello”}
	

[Document(…), ..]

on_prompt_start
	

[template_name]
		

{“question”: “hello”}
	

on_prompt_end
	

[template_name]
		

{“question”: “hello”}
	

ChatPromptValue(messages: [SystemMessage, …])

In addition to the standard events, users can also dispatch custom events (see example below).

Custom events will be only be surfaced with in the v2 version of the API!

A custom event has following format:

Attribute
	

Type
	

Description

name
	

str
	

A user defined name for the event.

data
	

Any
	

The data associated with the event. This can be anything, though we suggest making it JSON serializable.

Here are declarations associated with the standard events shown above:

format_docs:

def format_docs(docs: List[Document]) -> str:
    '''Format the docs.'''
    return ", ".join([doc.page_content for doc in docs])

format_docs = RunnableLambda(format_docs)

some_tool:

@tool
def some_tool(x: int, y: str) -> dict:
    '''Some_tool.'''
    return {"x": x, "y": y}

prompt:

template = ChatPromptTemplate.from_messages(
    [("system", "You are Cat Agent 007"), ("human", "{question}")]
).with_config({"run_name": "my_template", "tags": ["my_template"]})

Example:

from langchain_core.runnables import RunnableLambda

async def reverse(s: str) -> str:
    return s[::-1]

chain = RunnableLambda(func=reverse)

events = [
    event async for event in chain.astream_events("hello", version="v2")
]

# will produce the following events (run_id, and parent_ids
# has been omitted for brevity):
[
    {
        "data": {"input": "hello"},
        "event": "on_chain_start",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
    {
        "data": {"chunk": "olleh"},
        "event": "on_chain_stream",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
    {
        "data": {"output": "olleh"},
        "event": "on_chain_end",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
]

Example: Dispatch Custom Event

from langchain_core.callbacks.manager import (
    adispatch_custom_event,
)
from langchain_core.runnables import RunnableLambda, RunnableConfig
import asyncio


async def slow_thing(some_input: str, config: RunnableConfig) -> str:
    """Do something that takes a long time."""
    await asyncio.sleep(1) # Placeholder for some slow operation
    await adispatch_custom_event(
        "progress_event",
        {"message": "Finished step 1 of 3"},
        config=config # Must be included for python < 3.10
    )
    await asyncio.sleep(1) # Placeholder for some slow operation
    await adispatch_custom_event(
        "progress_event",
        {"message": "Finished step 2 of 3"},
        config=config # Must be included for python < 3.10
    )
    await asyncio.sleep(1) # Placeholder for some slow operation
    return "Done"

slow_thing = RunnableLambda(slow_thing)

async for event in slow_thing.astream_events("some_input", version="v2"):
    print(event)

    Parameters:

            input (Any) – The input to the Runnable.

            config (RunnableConfig | None) – The config to use for the Runnable.

            version (Literal['v1', 'v2']) – The version of the schema to use either v2 or v1. Users should use v2. v1 is for backwards compatibility and will be deprecated in 0.4.0. No default will be assigned until the API is stabilized. custom events will only be surfaced in v2.

            include_names (Sequence[str] | None) – Only include events from runnables with matching names.

            include_types (Sequence[str] | None) – Only include events from runnables with matching types.

            include_tags (Sequence[str] | None) – Only include events from runnables with matching tags.

            exclude_names (Sequence[str] | None) – Exclude events from runnables with matching names.

            exclude_types (Sequence[str] | None) – Exclude events from runnables with matching types.

            exclude_tags (Sequence[str] | None) – Exclude events from runnables with matching tags.

            kwargs (Any) – Additional keyword arguments to pass to the Runnable. These will be passed to astream_log as this implementation of astream_events is built on top of astream_log.

    Yields:

        An async stream of StreamEvents.
    Raises:

        NotImplementedError – If the version is not v1 or v2.
    Return type:

        AsyncIterator[StandardStreamEvent | CustomStreamEvent]

batch(inputs: list[Input], config: RunnableConfig | list[RunnableConfig] | None = None, *, return_exceptions: bool = False, **kwargs: Any | None) → list[Output]

    Default implementation runs invoke in parallel using a thread pool executor.

    The default implementation of batch works well for IO bound runnables.

    Subclasses should override this method if they can batch more efficiently; e.g., if the underlying Runnable uses an API which supports a batch mode.

    Parameters:

            inputs (list[Input])

            config (RunnableConfig | list[RunnableConfig] | None)

            return_exceptions (bool)

            kwargs (Any | None)

    Return type:

        list[Output]

batch_as_completed(inputs: Sequence[Input], config: RunnableConfig | Sequence[RunnableConfig] | None = None, *, return_exceptions: bool = False, **kwargs: Any | None) → Iterator[tuple[int, Output | Exception]]

    Run invoke in parallel on a list of inputs, yielding results as they complete.

    Parameters:

            inputs (Sequence[Input])

            config (RunnableConfig | Sequence[RunnableConfig] | None)

            return_exceptions (bool)

            kwargs (Any | None)

    Return type:

        Iterator[tuple[int, Output | Exception]]

bind(**kwargs: Any) → Runnable[Input, Output]

Bind arguments to a Runnable, returning a new Runnable.

Useful when a Runnable in a chain requires an argument that is not in the output of the previous Runnable or included in the user input.

Parameters:

    kwargs (Any) – The arguments to bind to the Runnable.
Returns:

    A new Runnable with the arguments bound.
Return type:

    Runnable[Input, Output]

Example:

from langchain_community.chat_models import ChatOllama
from langchain_core.output_parsers import StrOutputParser

llm = ChatOllama(model='llama2')

# Without bind.
chain = (
    llm
    | StrOutputParser()
)

chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two three four five.'

# With bind.
chain = (
    llm.bind(stop=["three"])
    | StrOutputParser()
)

chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two'

clean_search_query(query: str) → str
[source]

    Parameters:

        query (str)
    Return type:

        str

configurable_alternatives(which: ConfigurableField, *, default_key: str = 'default', prefix_keys: bool = False, **kwargs: Runnable[Input, Output] | Callable[[], Runnable[Input, Output]]) → RunnableSerializable

Configure alternatives for Runnables that can be set at runtime.

Parameters:

        which (ConfigurableField) – The ConfigurableField instance that will be used to select the alternative.

        default_key (str) – The default key to use if no alternative is selected. Defaults to “default”.

        prefix_keys (bool) – Whether to prefix the keys with the ConfigurableField id. Defaults to False.

        **kwargs (Runnable[Input, Output] | Callable[[], Runnable[Input, Output]]) – A dictionary of keys to Runnable instances or callables that return Runnable instances.

Returns:

    A new Runnable with the alternatives configured.
Return type:

    RunnableSerializable

from langchain_anthropic import ChatAnthropic
from langchain_core.runnables.utils import ConfigurableField
from langchain_openai import ChatOpenAI

model = ChatAnthropic(
    model_name="claude-3-sonnet-20240229"
).configurable_alternatives(
    ConfigurableField(id="llm"),
    default_key="anthropic",
    openai=ChatOpenAI()
)

# uses the default model ChatAnthropic
print(model.invoke("which organization created you?").content)

# uses ChatOpenAI
print(
    model.with_config(
        configurable={"llm": "openai"}
    ).invoke("which organization created you?").content
)

configurable_fields(**kwargs: ConfigurableField | ConfigurableFieldSingleOption | ConfigurableFieldMultiOption) → RunnableSerializable

Configure particular Runnable fields at runtime.

Parameters:

    **kwargs (ConfigurableField | ConfigurableFieldSingleOption | ConfigurableFieldMultiOption) – A dictionary of ConfigurableField instances to configure.
Returns:

    A new Runnable with the fields configured.
Return type:

    RunnableSerializable

from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI

model = ChatOpenAI(max_tokens=20).configurable_fields(
    max_tokens=ConfigurableField(
        id="output_token_number",
        name="Max tokens in the output",
        description="The maximum number of tokens in the output",
    )
)

# max_tokens = 20
print(
    "max_tokens_20: ",
    model.invoke("tell me something about chess").content
)

# max_tokens = 200
print("max_tokens_200: ", model.with_config(
    configurable={"output_token_number": 200}
    ).invoke("tell me something about chess").content
)

classmethod from_llm(vectorstore: ~langchain_core.vectorstores.base.VectorStore, llm: ~langchain_core.language_models.llms.BaseLLM, search: ~langchain_community.utilities.google_search.GoogleSearchAPIWrapper, prompt: ~langchain_core.prompts.base.BasePromptTemplate | None = None, num_search_results: int = 1, text_splitter: ~langchain_text_splitters.character.RecursiveCharacterTextSplitter = <langchain_text_splitters.character.RecursiveCharacterTextSplitter object>, trust_env: bool = False, allow_dangerous_requests: bool = False) → WebResearchRetriever
[source]

    Initialize from llm using default template.

    Parameters:

            vectorstore (VectorStore) – Vector store for storing web pages

            llm (BaseLLM) – llm for search question generation

            search (GoogleSearchAPIWrapper) – GoogleSearchAPIWrapper

            prompt (BasePromptTemplate | None) – prompt to generating search questions

            num_search_results (int) – Number of pages per Google search

            text_splitter (RecursiveCharacterTextSplitter) – Text splitter for splitting web pages into chunks

            trust_env (bool) – Whether to use the http_proxy/https_proxy env variables or check .netrc for proxy configuration

            allow_dangerous_requests (bool) – A flag to force users to acknowledge the risks of SSRF attacks when using this retriever

    Returns:

        WebResearchRetriever
    Return type:

        WebResearchRetriever

get_relevant_documents(query: str, *, callbacks: Callbacks = None, tags: list[str] | None = None, metadata: dict[str, Any] | None = None, run_name: str | None = None, **kwargs: Any) → list[Document]

    Deprecated since version langchain-core==0.1.46: Use invoke() instead.

    Retrieve documents relevant to a query.

    Users should favor using .invoke or .batch rather than get_relevant_documents directly.

    Parameters:

            query (str) – string to find relevant documents for.

            callbacks (Callbacks) – Callback manager or list of callbacks. Defaults to None.

            tags (Optional[list[str]]) – Optional list of tags associated with the retriever. These tags will be associated with each call to this retriever, and passed as arguments to the handlers defined in callbacks. Defaults to None.

            metadata (Optional[dict[str, Any]]) – Optional metadata associated with the retriever. This metadata will be associated with each call to this retriever, and passed as arguments to the handlers defined in callbacks. Defaults to None.

            run_name (Optional[str]) – Optional name for the run. Defaults to None.

            kwargs (Any) – Additional arguments to pass to the retriever.

    Returns:

        List of relevant documents.
    Return type:

        list[Document]

invoke(input: str, config: RunnableConfig | None = None, **kwargs: Any) → list[Document]

Invoke the retriever to get relevant documents.

Main entry point for synchronous retriever invocations.

Parameters:

        input (str) – The query string.

        config (RunnableConfig | None) – Configuration for the retriever. Defaults to None.

        kwargs (Any) – Additional arguments to pass to the retriever.

Returns:

    List of relevant documents.
Return type:

    list[Document]

Examples:

retriever.invoke("query")

search_tool(query: str, num_search_results: int = 1) → List[dict]
[source]

    Returns num_search_results pages per Google search.

    Parameters:

            query (str)

            num_search_results (int)

    Return type:

        List[dict]

stream(input: Input, config: RunnableConfig | None = None, **kwargs: Any | None) → Iterator[Output]

    Default implementation of stream, which calls invoke. Subclasses should override this method if they support streaming output.

    Parameters:

            input (Input) – The input to the Runnable.

            config (RunnableConfig | None) – The config to use for the Runnable. Defaults to None.

            kwargs (Any | None) – Additional keyword arguments to pass to the Runnable.

    Yields:

        The output of the Runnable.
    Return type:

        Iterator[Output]

with_alisteners(*, on_start: AsyncListener | None = None, on_end: AsyncListener | None = None, on_error: AsyncListener | None = None) → Runnable[Input, Output]

Bind asynchronous lifecycle listeners to a Runnable, returning a new Runnable.

on_start: Asynchronously called before the Runnable starts running. on_end: Asynchronously called after the Runnable finishes running. on_error: Asynchronously called if the Runnable throws an error.

The Run object contains information about the run, including its id, type, input, output, error, start_time, end_time, and any tags or metadata added to the run.

Parameters:

        on_start (Optional[AsyncListener]) – Asynchronously called before the Runnable starts running. Defaults to None.

        on_end (Optional[AsyncListener]) – Asynchronously called after the Runnable finishes running. Defaults to None.

        on_error (Optional[AsyncListener]) – Asynchronously called if the Runnable throws an error. Defaults to None.

Returns:

    A new Runnable with the listeners bound.
Return type:

    Runnable[Input, Output]

Example:

from langchain_core.runnables import RunnableLambda
import time

async def test_runnable(time_to_sleep : int):
    print(f"Runnable[{time_to_sleep}s]: starts at {format_t(time.time())}")
    await asyncio.sleep(time_to_sleep)
    print(f"Runnable[{time_to_sleep}s]: ends at {format_t(time.time())}")

async def fn_start(run_obj : Runnable):
    print(f"on start callback starts at {format_t(time.time())}
    await asyncio.sleep(3)
    print(f"on start callback ends at {format_t(time.time())}")

async def fn_end(run_obj : Runnable):
    print(f"on end callback starts at {format_t(time.time())}
    await asyncio.sleep(2)
    print(f"on end callback ends at {format_t(time.time())}")

runnable = RunnableLambda(test_runnable).with_alisteners(
    on_start=fn_start,
    on_end=fn_end
)
async def concurrent_runs():
    await asyncio.gather(runnable.ainvoke(2), runnable.ainvoke(3))

asyncio.run(concurrent_runs())
Result:
on start callback starts at 2024-05-16T14:20:29.637053+00:00
on start callback starts at 2024-05-16T14:20:29.637150+00:00
on start callback ends at 2024-05-16T14:20:32.638305+00:00
on start callback ends at 2024-05-16T14:20:32.638383+00:00
Runnable[3s]: starts at 2024-05-16T14:20:32.638849+00:00
Runnable[5s]: starts at 2024-05-16T14:20:32.638999+00:00
Runnable[3s]: ends at 2024-05-16T14:20:35.640016+00:00
on end callback starts at 2024-05-16T14:20:35.640534+00:00
Runnable[5s]: ends at 2024-05-16T14:20:37.640169+00:00
on end callback starts at 2024-05-16T14:20:37.640574+00:00
on end callback ends at 2024-05-16T14:20:37.640654+00:00
on end callback ends at 2024-05-16T14:20:39.641751+00:00

with_config(config: RunnableConfig | None = None, **kwargs: Any) → Runnable[Input, Output]

    Bind config to a Runnable, returning a new Runnable.

    Parameters:

            config (RunnableConfig | None) – The config to bind to the Runnable.

            kwargs (Any) – Additional keyword arguments to pass to the Runnable.

    Returns:

        A new Runnable with the config bound.
    Return type:

        Runnable[Input, Output]

with_fallbacks(fallbacks: Sequence[Runnable[Input, Output]], *, exceptions_to_handle: tuple[type[BaseException], ...] = (<class 'Exception'>,), exception_key: Optional[str] = None) → RunnableWithFallbacksT[Input, Output]

Add fallbacks to a Runnable, returning a new Runnable.

The new Runnable will try the original Runnable, and then each fallback in order, upon failures.

Parameters:

        fallbacks (Sequence[Runnable[Input, Output]]) – A sequence of runnables to try if the original Runnable fails.

        exceptions_to_handle (tuple[type[BaseException], ...]) – A tuple of exception types to handle. Defaults to (Exception,).

        exception_key (Optional[str]) – If string is specified then handled exceptions will be passed to fallbacks as part of the input under the specified key. If None, exceptions will not be passed to fallbacks. If used, the base Runnable and its fallbacks must accept a dictionary as input. Defaults to None.

Returns:

    A new Runnable that will try the original Runnable, and then each fallback in order, upon failures.
Return type:

    RunnableWithFallbacksT[Input, Output]

Example

from typing import Iterator

from langchain_core.runnables import RunnableGenerator


def _generate_immediate_error(input: Iterator) -> Iterator[str]:
    raise ValueError()
    yield ""


def _generate(input: Iterator) -> Iterator[str]:
    yield from "foo bar"


runnable = RunnableGenerator(_generate_immediate_error).with_fallbacks(
    [RunnableGenerator(_generate)]
    )
print(''.join(runnable.stream({}))) #foo bar

    Parameters:

            fallbacks (Sequence[Runnable[Input, Output]]) – A sequence of runnables to try if the original Runnable fails.

            exceptions_to_handle (tuple[type[BaseException], ...]) – A tuple of exception types to handle.

            exception_key (Optional[str]) – If string is specified then handled exceptions will be passed to fallbacks as part of the input under the specified key. If None, exceptions will not be passed to fallbacks. If used, the base Runnable and its fallbacks must accept a dictionary as input.

    Returns:

        A new Runnable that will try the original Runnable, and then each fallback in order, upon failures.
    Return type:

        RunnableWithFallbacksT[Input, Output]

with_listeners(*, on_start: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None = None, on_end: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None = None, on_error: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None = None) → Runnable[Input, Output]

Bind lifecycle listeners to a Runnable, returning a new Runnable.

on_start: Called before the Runnable starts running, with the Run object. on_end: Called after the Runnable finishes running, with the Run object. on_error: Called if the Runnable throws an error, with the Run object.

The Run object contains information about the run, including its id, type, input, output, error, start_time, end_time, and any tags or metadata added to the run.

Parameters:

        on_start (Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]]) – Called before the Runnable starts running. Defaults to None.

        on_end (Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]]) – Called after the Runnable finishes running. Defaults to None.

        on_error (Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]]) – Called if the Runnable throws an error. Defaults to None.

Returns:

    A new Runnable with the listeners bound.
Return type:

    Runnable[Input, Output]

Example:

from langchain_core.runnables import RunnableLambda
from langchain_core.tracers.schemas import Run

import time

def test_runnable(time_to_sleep : int):
    time.sleep(time_to_sleep)

def fn_start(run_obj: Run):
    print("start_time:", run_obj.start_time)

def fn_end(run_obj: Run):
    print("end_time:", run_obj.end_time)

chain = RunnableLambda(test_runnable).with_listeners(
    on_start=fn_start,
    on_end=fn_end
)
chain.invoke(2)

with_retry(*, retry_if_exception_type: tuple[type[BaseException], ...] = (<class 'Exception'>,), wait_exponential_jitter: bool = True, stop_after_attempt: int = 3) → Runnable[Input, Output]

Create a new Runnable that retries the original Runnable on exceptions.

Parameters:

        retry_if_exception_type (tuple[type[BaseException], ...]) – A tuple of exception types to retry on. Defaults to (Exception,).

        wait_exponential_jitter (bool) – Whether to add jitter to the wait time between retries. Defaults to True.

        stop_after_attempt (int) – The maximum number of attempts to make before giving up. Defaults to 3.

Returns:

    A new Runnable that retries the original Runnable on exceptions.
Return type:

    Runnable[Input, Output]

Example:

from langchain_core.runnables import RunnableLambda

count = 0


def _lambda(x: int) -> None:
    global count
    count = count + 1
    if x == 1:
        raise ValueError("x is 1")
    else:
         pass


runnable = RunnableLambda(_lambda)
try:
    runnable.with_retry(
        stop_after_attempt=2,
        retry_if_exception_type=(ValueError,),
    ).invoke(1)
except ValueError:
    pass

assert (count == 2)

    Parameters:

            retry_if_exception_type (tuple[type[BaseException], ...]) – A tuple of exception types to retry on

            wait_exponential_jitter (bool) – Whether to add jitter to the wait time between retries

            stop_after_attempt (int) – The maximum number of attempts to make before giving up

    Returns:

        A new Runnable that retries the original Runnable on exceptions.
    Return type:

        Runnable[Input, Output]

with_types(*, input_type: type[Input] | None = None, output_type: type[Output] | None = None) → Runnable[Input, Output]

    Bind input and output types to a Runnable, returning a new Runnable.

    Parameters:

            input_type (type[Input] | None) – The input type to bind to the Runnable. Defaults to None.

            output_type (type[Output] | None) – The output type to bind to the Runnable. Defaults to None.

    Returns:

        A new Runnable with the types bound.
    Return type:

        Runnable[Input, Output]
















and integrate that into my code. respond with the full working code:
import os
import json
import shutil
import gradio as gr
from itertools import chain

# Import necessary modules
import arxiv
from langchain.chains import LLMChain
from langchain_core.prompts import PromptTemplate, ChatPromptTemplate
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_community.embeddings.sentence_transformer import SentenceTransformerEmbeddings
from langchain_unstructured import UnstructuredLoader
from langchain_text_splitters import CharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_community.retrievers import ArxivRetriever
from langchain.document_loaders import PyPDFLoader
from langchain.document_loaders import DirectoryLoader, UnstructuredURLLoader
from langchain.tools import DuckDuckGoSearchRun
from langchain_ollama import ChatOllama

# For LangSmith tracing
from langsmith import traceable
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "<TRUFFLEHOG_REDACTED_LANGSMITH>"  # Replace with your actual API key
os.environ["LANGCHAIN_PROJECT"] = "l"  # Optional: "default" is used if not set

# Initialize embeddings
embeddings = SentenceTransformerEmbeddings(model_name="all-MiniLM-L6-v2")

# Knowledge Base Vector Store
persist_directory = "./chroma_db"
vectorstore = Chroma(persist_directory=persist_directory, embedding_function=embeddings)

# arXiv Vector Store (Separate)
arxiv_persist_directory = "./arxiv_chroma_db"
arxiv_vectorstore = Chroma(persist_directory=arxiv_persist_directory, embedding_function=embeddings)

# Ensure the 'uploads' directories exist
if not os.path.exists("uploads"):
    os.makedirs("uploads")
if not os.path.exists("arxiv_uploads"):
    os.makedirs("arxiv_uploads")

# Function to load or initialize knowledge base JSON
def load_knowledge_base_json():
    if os.path.exists("knowledge_base.json"):
        with open("knowledge_base.json", "r") as f:
            knowledge_base_data = json.load(f)
    else:
        knowledge_base_data = {"files": [], "urls": [], "arxiv_papers": [], "arxiv_files": []}
    return knowledge_base_data

knowledge_base_data = load_knowledge_base_json()

# Function to save knowledge base JSON
def save_knowledge_base_json():
    with open("knowledge_base.json", "w") as f:
        json.dump(knowledge_base_data, f)

# Function to get processed files
def get_processed_files():
    files = os.listdir("uploads")
    return "### Processed Files:\n" + "\n".join(files) if files else "No files processed yet."

# Function to display processed files and URLs
def display_processed_items():
    global knowledge_base_data
    files_text = get_processed_files()
    urls = knowledge_base_data.get("urls", [])
    display_text = files_text + "\n\n### Processed URLs:\n"
    if urls:
        display_text += "\n".join(urls)
    else:
        display_text += "No URLs processed yet."
    return display_text

# Function to update the knowledge base
@traceable
def update_knowledge_base(uploaded_files, website_urls):
    global knowledge_base_data
    if not uploaded_files and not website_urls:
        return "No new documents or URLs to process.", display_processed_items()

    processed_files = []
    skipped_files = []
    docs = []
    new_urls = []

    # Process uploaded files
    if uploaded_files:
        for file in uploaded_files:
            file_path = file.name
            destination_path = os.path.join("uploads", os.path.basename(file_path))
            file_name = os.path.basename(file_path)

            if os.path.exists(destination_path) or file_name in knowledge_base_data["files"]:
                skipped_files.append(file_name)
            else:
                try:
                    shutil.move(file_path, destination_path)
                    processed_files.append(file_name)
                except Exception as e:
                    print(f"Error moving file {file_name}: {str(e)}")

    # Load documents from uploaded files
    if processed_files:
        loader = DirectoryLoader("uploads", glob="**/*")
        documents = loader.load()
        docs.extend(documents)

    # Load documents from URLs
    if website_urls:
        urls = website_urls.split(',')
        for url in urls:
            url = url.strip()
            if url and url not in knowledge_base_data["urls"]:
                loader = UnstructuredURLLoader(urls=[url])
                try:
                    loaded_docs = loader.load()
                    docs.extend(loaded_docs)
                    new_urls.append(url)
                except Exception as e:
                    print(f"Error loading URL {url}: {str(e)}")
            else:
                skipped_files.append(url)

    # Split documents and add to vector store
    if docs:
        text_splitter = CharacterTextSplitter(chunk_size=2500, chunk_overlap=200, length_function=len)
        split_docs = text_splitter.split_documents(docs)
        vectorstore.add_documents(split_docs)
        vectorstore.persist()
        # Update knowledge base data and save to JSON
        knowledge_base_data["files"].extend(processed_files)
        knowledge_base_data["urls"].extend(new_urls)
        save_knowledge_base_json()
        status_message = f"Processed {len(processed_files)} new files and {len(new_urls)} new URLs."
        if skipped_files:
            status_message += f"\nSkipped existing files/URLs: {', '.join(skipped_files)}"
    else:
        status_message = "No new documents or URLs were added."
    return status_message, display_processed_items()

# Function to clear database and uploaded files
def clear_database():
    global vectorstore, knowledge_base_data

    try:
        vectorstore._client._client.close()
    except AttributeError:
        pass

    shutil.rmtree("uploads", ignore_errors=True)
    os.makedirs("uploads", exist_ok=True)

    shutil.rmtree(persist_directory, ignore_errors=True)

    vectorstore = Chroma(persist_directory=persist_directory, embedding_function=embeddings)

    # Clear the knowledge base data
    knowledge_base_data = {"files": [], "urls": [], "arxiv_papers": [], "arxiv_files": []}
    # Remove the JSON file
    if os.path.exists("knowledge_base.json"):
        os.remove("knowledge_base.json")

    return "Knowledge base and uploaded files cleared.", display_processed_items()

# Function to process uploaded files in the chat and include their content
def process_uploaded_files(files):
    content = ""
    if files:
        for file in files:
            if file.name.endswith((".pdf", ".txt", ".docx")):
                loader = UnstructuredLoader(file_path=file.name)
                try:
                    docs = loader.load()
                    content += "\n".join([doc.page_content for doc in docs])
                except Exception as e:
                    print(f"Error processing file {file.name}: {str(e)}")
    return content

# Function to count tokens
def count_tokens(text):
    return len(text.split())  # Simple token count; adjust with tokenizer if needed

# Initialize Llama model
llm = ChatOllama(
    model="llama3.1",
    temperature=0,
    base_url="http://localhost:11434",
    streaming=True
)

# Maximum token limit
MAX_TOKENS = 128000  # Example maximum token limit

# Chat with Llama (normal chat) function
@traceable
def chat_with_llama(message, history, uploaded_files):
    # Process uploaded files and include their content
    file_content = process_uploaded_files(uploaded_files)
    # Convert history to list of messages
    messages = [SystemMessage(content="You are a helpful assistant.")]
    if history:
        for user_msg, assistant_msg in history:
            if user_msg:
                messages.append(HumanMessage(content=user_msg))
                if assistant_msg is not None:
                    messages.append(AIMessage(content=assistant_msg))
    # Append file content to user message if any
    if file_content:
        message += f"\n\nContent files Uploaded:\n{file_content}"
        message += "\n\nWith that knowledge answer to the question as best as possible: "
    messages.append(HumanMessage(content=message))
    # Check if the total tokens exceed the maximum allowed tokens
    total_tokens = sum(count_tokens(msg.content) for msg in messages)
    if total_tokens > MAX_TOKENS:
        # Truncate the history to fit within the token limit
        while total_tokens > MAX_TOKENS and len(messages) > 2:
            # Remove the oldest user-assistant message pair
            messages.pop(1)
            messages.pop(1)
            total_tokens = sum(count_tokens(msg.content) for msg in messages)
        if total_tokens > MAX_TOKENS:
            return "Error: The conversation exceeds the maximum token limit. Please start a new chat."
    # Stream the response from the model
    try:
        response_stream = llm.stream(messages)
        partial_response = ""
        for chunk in response_stream:
            partial_response += chunk.content
            yield partial_response
    except Exception as e:
        yield f"Error: {str(e)}"

# Function to generate search queries
def generate_search_queries(message):
    query_generation_prompt = PromptTemplate(
        input_variables=["user_message"],
        template="""
Given the user's message:
"{user_message}"

Generate two distinct search queries that would help retrieve the most relevant documents from the knowledge base to answer the user's question. Be very simple. For a question like "What is the capital of France?" just do a query with "capital" and "France capital".

Provide the search queries as a JSON list, for example:
["search query 1", "search query 2"]
"""
    )
    llm_chain = LLMChain(llm=llm, prompt=query_generation_prompt)
    search_queries_json = llm_chain.run(user_message=message)
    # Parse the JSON output
    try:
        search_queries = json.loads(search_queries_json)
    except json.JSONDecodeError:
        # If parsing fails, default to using the original message
        search_queries = [message]
    if not isinstance(search_queries, list):
        search_queries = [search_queries]
    return search_queries

# Updated Function to handle knowledge base Q&A
@traceable
def knowledge_base_chat(message, history):
    # Generate search queries
    search_queries = generate_search_queries(message)
    # Perform retrieval for each generated query
    all_results = []
    for query in search_queries:
        results = vectorstore.similarity_search(query, k=4)
        all_results.extend(results)
    if not all_results:
        yield "No relevant information found in the knowledge base."
        return
    # Remove duplicate documents
    unique_results = list({doc.page_content: doc for doc in all_results}.values())
    # Use the LLM to extract important information
    extraction_prompt = PromptTemplate(
        input_variables=["user_message", "document"],
        template="""
Given the user's message:
"{user_message}"

And the following document:

{document}

Extract important information that could help to answer the question. Respond with the relevant parts of the document in original text, omitting irrelevant sections.
"""
    )
    extracted_info_list = []
    extraction_chain = LLMChain(llm=llm, prompt=extraction_prompt)
    for doc in unique_results:
        extracted_info = extraction_chain.run(user_message=message, document=doc.page_content)
        extracted_info_list.append(extracted_info)
    # Combine the extracted information
    context = "\n\n".join(extracted_info_list)
    # Build the conversation history
    messages = [SystemMessage(content="You are a helpful assistant that answers questions based on the provided context.")]
    if history:
        for user_msg, assistant_msg in history:
            if user_msg:
                messages.append(HumanMessage(content=user_msg))
                if assistant_msg is not None:
                    messages.append(AIMessage(content=assistant_msg))
    # Add the latest message with context
    prompt = f"Context:\n{context}\n\nQuestion:\n{message}\n\nAnswer:"
    messages.append(HumanMessage(content=prompt))
    # Stream the response from the model
    try:
        response_stream = llm.stream(messages)
        partial_response = ""
        for chunk in response_stream:
            partial_response += chunk.content
            yield partial_response
    except Exception as e:
        yield f"Error: {str(e)}"

# Function for the Internet-Enabled LLM
@traceable
def internet_access_llm(message, history, use_wikipedia):
    # Generate a query from the user input
    query = message
    # Search the internet
    search_tool = DuckDuckGoSearchRun()
    search_results = search_tool.run(query)
    # Optionally search Wikipedia
    if use_wikipedia:
        wikipedia_tool = DuckDuckGoSearchRun(backend="wikipedia")
        wiki_results = wikipedia_tool.run(query)
        search_results += "\n\n" + wiki_results
    # Build the conversation history
    messages = [SystemMessage(content="You are an assistant that uses internet search results to provide accurate answers.")]
    if history:
        for user_msg, assistant_msg in history:
            if user_msg:
                messages.append(HumanMessage(content=user_msg))
                if assistant_msg is not None:
                    messages.append(AIMessage(content=assistant_msg))
    # Add the latest message with search results
    prompt = f"Search Results:\n{search_results}\n\nQuestion:\n{message}\n\nAnswer:"
    messages.append(HumanMessage(content=prompt))
    # Stream the response from the model
    try:
        response_stream = llm.stream(messages)
        partial_response = ""
        for chunk in response_stream:
            partial_response += chunk.content
            yield partial_response
    except Exception as e:
        yield f"Error: {str(e)}"

# Function to retrieve arXiv papers and update arXiv knowledge base
@traceable
def arxiv_research(arxiv_query, max_papers, uploaded_files, arxiv_ids_or_links):
    if not arxiv_query and not uploaded_files and not arxiv_ids_or_links:
        return "No query, files, or IDs provided.", display_arxiv_items()

    processed_papers = []
    skipped_papers = []
    docs = []

    # Search arXiv using the query
    if arxiv_query:
        search = arxiv.Search(query=arxiv_query, max_results=int(max_papers))
        for result in search.results():
            paper_id = result.get_short_id()
            paper_title = result.title
            pdf_url = result.pdf_url
            # Check if paper is already processed
            if paper_id in knowledge_base_data.get("arxiv_papers", []):
                skipped_papers.append(paper_title)
                continue
            # Download PDF
            try:
                paper = next(arxiv.Client().results(arxiv.Search(id_list=[paper_id])))
                pdf_path = paper.download_pdf(dirpath="arxiv_uploads")
                # Extract text
                loader = PyPDFLoader(pdf_path)
                paper_doc = loader.load()
                docs.extend(paper_doc)
                processed_papers.append(paper_title)
                # Save paper info
                knowledge_base_data.setdefault("arxiv_papers", []).append(paper_id)
            except Exception as e:
                print(f"Error downloading or processing paper {paper_title}: {str(e)}")

    # Process uploaded files
    if uploaded_files:
        for file in uploaded_files:
            file_path = file.name
            destination_path = os.path.join("arxiv_uploads", os.path.basename(file_path))
            file_name = os.path.basename(file_path)
            if os.path.exists(destination_path) or file_name in knowledge_base_data.get("arxiv_files", []):
                skipped_papers.append(file_name)
            else:
                try:
                    shutil.move(file_path, destination_path)
                    loader = PyPDFLoader(destination_path)
                    paper_doc = loader.load()
                    docs.extend(paper_doc)
                    processed_papers.append(file_name)
                    knowledge_base_data.setdefault("arxiv_files", []).append(file_name)
                except Exception as e:
                    print(f"Error processing file {file_name}: {str(e)}")

    # Process arXiv IDs or URLs
    if arxiv_ids_or_links:
        ids_or_links = arxiv_ids_or_links.split(',')
        for item in ids_or_links:
            item = item.strip()
            if not item:
                continue
            # Extract arXiv ID
            if 'arxiv.org' in item:
                paper_id = item.split('/')[-1]
            else:
                paper_id = item
            if paper_id in knowledge_base_data.get("arxiv_papers", []):
                skipped_papers.append(paper_id)
                continue
            # Download and process paper
            try:
                paper = next(arxiv.Client().results(arxiv.Search(id_list=[paper_id])))
                pdf_path = paper.download_pdf(dirpath="arxiv_uploads")
                loader = PyPDFLoader(pdf_path)
                paper_doc = loader.load()
                docs.extend(paper_doc)
                processed_papers.append(paper.title)
                knowledge_base_data.setdefault("arxiv_papers", []).append(paper_id)
            except Exception as e:
                print(f"Error processing paper {paper_id}: {str(e)}")

    # Split documents and add to arXiv vector store
    if docs:
        text_splitter = CharacterTextSplitter(chunk_size=2500, chunk_overlap=200, length_function=len)
        split_docs = text_splitter.split_documents(docs)
        arxiv_vectorstore.add_documents(split_docs)
        arxiv_vectorstore.persist()
        # Save knowledge base data
        save_knowledge_base_json()
        status_message = f"Processed {len(processed_papers)} new papers."
        if skipped_papers:
            status_message += f"\nSkipped existing papers: {', '.join(skipped_papers)}"
    else:
        status_message = "No new papers were added."
    return status_message, display_arxiv_items()

# Function to display processed arXiv items
def display_arxiv_items():
    arxiv_files = os.listdir("arxiv_uploads")
    arxiv_papers = knowledge_base_data.get("arxiv_papers", [])
    display_text = "### Processed arXiv Papers:\n"
    if arxiv_papers:
        display_text += "\n".join(arxiv_papers)
    else:
        display_text += "No arXiv papers processed yet."
    display_text += "\n\n### Uploaded Files:\n"
    if arxiv_files:
        display_text += "\n".join(arxiv_files)
    else:
        display_text += "No files uploaded yet."
    return display_text

# Function to allow users to download papers
def download_papers():
    files = os.listdir("arxiv_uploads")
    file_paths = [os.path.join("arxiv_uploads", f) for f in files]
    return file_paths

# Function to handle arXiv Knowledge Base Q&A
@traceable
def ask_arxiv_knowledge_base(message, history):
    # Generate search queries
    search_queries = generate_search_queries(message)
    # Perform retrieval
    all_results = []
    for query in search_queries:
        results = arxiv_vectorstore.similarity_search(query, k=4)
        all_results.extend(results)
    if not all_results:
        yield "No relevant information found in the arXiv knowledge base."
        return
    # Remove duplicates
    unique_results = list({doc.page_content: doc for doc in all_results}.values())
    # Use LLM to extract information
    extraction_prompt = PromptTemplate(
        input_variables=["user_message", "document"],
        template="""
Given the user's message:
"{user_message}"

And the following document:

{document}

Extract important information that could help to answer the question. Respond with the relevant parts of the document in original text, omitting irrelevant sections.
"""
    )
    extracted_info_list = []
    extraction_chain = LLMChain(llm=llm, prompt=extraction_prompt)
    for doc in unique_results:
        extracted_info = extraction_chain.run(user_message=message, document=doc.page_content)
        extracted_info_list.append(extracted_info)
    # Combine the extracted information
    context = "\n\n".join(extracted_info_list)
    # Also prepare a list of papers used
    papers_used = []
    for doc in unique_results:
        metadata = doc.metadata
        title = metadata.get('Title', 'Unknown title')
        authors = metadata.get('Authors', 'Unknown authors')
        paper_info = f"Title: {title}, Authors: {authors}"
        papers_used.append(paper_info)
    papers_info = "\n".join([f"- {info}" for info in papers_used])
    # Build the conversation history
    messages = [SystemMessage(content="You are a helpful assistant that answers questions based on arXiv research papers.")]
    if history:
        for user_msg, assistant_msg in history:
            if user_msg:
                messages.append(HumanMessage(content=user_msg))
                if assistant_msg is not None:
                    messages.append(AIMessage(content=assistant_msg))
    # Add the latest message with context
    prompt = f"The following arXiv papers were used:\n{papers_info}\n\nContext:\n{context}\n\nQuestion:\n{message}\n\nAnswer:"
    messages.append(HumanMessage(content=prompt))
    # Stream the response from the model
    try:
        response_stream = llm.stream(messages)
        partial_response = ""
        for chunk in response_stream:
            partial_response += chunk.content
            yield partial_response
    except Exception as e:
        yield f"Error: {str(e)}"

# Initialize the ArxivRetriever
arxiv_retriever = ArxivRetriever(
    load_max_docs=5,
    get_full_documents=True,
)

# Function for the Arxiv Retriever Chat
@traceable
def arxiv_retriever_chat(message, history):
    # Use the ArxivRetriever to get documents
    docs = arxiv_retriever.invoke(message)

    if not docs:
        yield "No relevant papers found."
        return

    # Prepare context from docs
    context = "\n\n".join([doc.page_content for doc in docs])

    # Also prepare a list of papers used
    papers_used = []
    for doc in docs:
        title = doc.metadata.get('Title', 'Unknown title')
        authors = doc.metadata.get('Authors', 'Unknown authors')
        paper_info = f"Title: {title}, Authors: {authors}"
        papers_used.append(paper_info)
    papers_info = "\n".join([f"- {info}" for info in papers_used])

    # Build the conversation history
    messages = [SystemMessage(content="You are a helpful assistant that answers questions based on arXiv papers.")]

    if history:
        for user_msg, assistant_msg in history:
            if user_msg:
                messages.append(HumanMessage(content=user_msg))
                if assistant_msg is not None:
                    messages.append(AIMessage(content=assistant_msg))

    # Add the latest message with context
    prompt = f"The following arXiv papers were used:\n{papers_info}\n\nContext:\n{context}\n\nQuestion:\n{message}\n\nAnswer:"

    messages.append(HumanMessage(content=prompt))

    # Stream the response from the model
    try:
        response_stream = llm.stream(messages)
        partial_response = ""
        for chunk in response_stream:
            partial_response += chunk.content
            yield partial_response
    except Exception as e:
        yield f"Error: {str(e)}"

# Gradio interface
with gr.Blocks() as demo:
    gr.Markdown("# Multi-Tab Gradio App with Llama and Similarity Search")

    # Chat with Llama Tab
    with gr.Tab("Chat with Llama"):
        gr.Markdown("## Chat Interface")
        with gr.Row():
            upload_files = gr.File(
                label="Upload PDFs, DOCX or TXT files",
                file_types=[".pdf", ".txt", ".docx"],
                file_count="multiple",
                visible=True
            )
        gr_chat = gr.ChatInterface(
            fn=chat_with_llama,
            title="Chat with Llama",
            additional_inputs=[upload_files]
        )

    # Knowledge Base Tab
    with gr.Tab("Knowledge Base"):
        gr.Markdown("## Knowledge Base Management and Chat")
        with gr.Row():
            with gr.Column(scale=1):
                gr.Markdown("### Update Knowledge Base")
                kb_upload_files = gr.File(
                    label="Upload Files",
                    file_types=[".pdf", ".txt", ".docx"],
                    file_count="multiple"
                )
                kb_website_urls = gr.Textbox(
                    label="Website URLs",
                    placeholder="Enter comma-separated URLs"
                )
                kb_update_button = gr.Button("Update Knowledge Base")
                kb_clear_button = gr.Button("Clear Database")
                kb_update_status = gr.Markdown()
                kb_update_button.click(
                    update_knowledge_base,
                    inputs=[kb_upload_files, kb_website_urls],
                    outputs=[kb_update_status, kb_update_status]
                )
                kb_clear_button.click(
                    clear_database,
                    outputs=[kb_update_status, kb_update_status]
                )
                gr.Markdown("### Processed Files and URLs")
                processed_items_display = gr.Markdown(display_processed_items())
                # Update the display when the knowledge base is updated or cleared
                kb_update_button.click(
                    update_knowledge_base,
                    inputs=[kb_upload_files, kb_website_urls],
                    outputs=[kb_update_status, processed_items_display]
                )
                kb_clear_button.click(
                    clear_database,
                    outputs=[kb_update_status, processed_items_display]
                )
            with gr.Column(scale=2):
                gr.Markdown("### Chat with Knowledge Base")
                gr_chat_kb = gr.ChatInterface(
                    fn=knowledge_base_chat,
                    title="Knowledge Base Chat"
                )

    # LLM with Internet Access Tab
    with gr.Tab("LLM with Internet Access"):
        gr.Markdown("## Internet-Enabled LLM")
        use_wikipedia = gr.Checkbox(label="Use Wikipedia")
        gr_chat_internet = gr.ChatInterface(
            fn=internet_access_llm,
            title="LLM with Internet Access",
            additional_inputs=[use_wikipedia]
        )

    # arXiv Research Papers Tab
    with gr.Tab("arXiv Research Papers"):
        gr.Markdown("## Retrieve arXiv Papers and Chat")
        with gr.Row():
            with gr.Column(scale=1):
                gr.Markdown("### Retrieve/Add arXiv Papers")
                arxiv_query = gr.Textbox(
                    label="Search Query",
                    placeholder="Enter your arXiv search query"
                )
                max_papers = gr.Slider(
                    label="Number of Papers",
                    minimum=1,
                    maximum=10,
                    step=1,
                    value=5
                )
                arxiv_upload_files = gr.File(
                    label="Upload PDF Files",
                    file_types=[".pdf"],
                    file_count="multiple"
                )
                arxiv_ids_or_links = gr.Textbox(
                    label="arXiv IDs or URLs",
                    placeholder="Enter comma-separated arXiv IDs or URLs"
                )
                arxiv_button = gr.Button("Retrieve/Add Papers")
                arxiv_result = gr.Markdown()
                arxiv_button.click(
                    arxiv_research,
                    inputs=[arxiv_query, max_papers, arxiv_upload_files, arxiv_ids_or_links],
                    outputs=[arxiv_result, arxiv_result]
                )
                gr.Markdown("### Processed arXiv Papers")
                processed_items_display_arxiv = gr.Markdown(display_arxiv_items())
                # Update the display when new papers are added
                arxiv_button.click(
                    arxiv_research,
                    inputs=[arxiv_query, max_papers, arxiv_upload_files, arxiv_ids_or_links],
                    outputs=[arxiv_result, processed_items_display_arxiv]
                )
                gr.Markdown("### Download Papers")
                download_button = gr.Button("Download Papers")
                download_files = gr.File()
                download_button.click(
                    download_papers,
                    outputs=[download_files]
                )
            with gr.Column(scale=2):
                gr.Markdown("### Chat with arXiv Knowledge Base")
                gr_chat_arxiv = gr.ChatInterface(
                    fn=ask_arxiv_knowledge_base,
                    title="arXiv Knowledge Base Chat"
                )

    # New Arxiv Retriever Tab
    with gr.Tab("Arxiv Retriever"):
        gr.Markdown("## Ask Questions Based on arXiv Papers")
        gr.Markdown("This tab uses the ArxivRetriever to fetch relevant papers and provide answers.")
        gr_chat_arxiv_retriever = gr.ChatInterface(
            fn=arxiv_retriever_chat,
            title="Arxiv Retriever Chat"
        )

    # Launch the app
    demo.launch()