Hi! I’ve been trying to use the Langchain library in Palantir. However, the only method I’ve found requires the use of an personal Open API key, using external resource method in Palantir.
Is there a way to utilize our company’s Palantir AIP tokens instead?
3 Likes
To use langchain in a transform, you could use a Palantir-provided model in a transform and wrap that in a small wrapper in langchain: https://python.langchain.com/v0.2/docs/how_to/custom_llm/
Thank you for your feedback.
I tried my best but only palantir_llm.invoke(question) option I made is feasible.
My purpose of applying langchain is that using various option including pipeline but not working with small wrapping.
from transforms.api import transform, Input, Output
from palantir_models.transforms import OpenAiGptChatLanguageModelInput
from palantir_models.models import OpenAiGptChatLanguageModel
from language_model_service_api.languagemodelservice_api_completion_v3 import GptChatCompletionRequest
from language_model_service_api.languagemodelservice_api import ChatMessage, ChatMessageRole
import pandas as pd
import logging
from typing import Optional, List, Any, Tuple
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class LLM:
"""Base class for Language Models."""
def _call(self, prompt: str, stop: Optional[List[str]] = None, **kwargs: Any) -> str:
raise NotImplementedError
@property
def _llm_type(self) -> str:
raise NotImplementedError
class PalantirLLM(LLM):
"""A custom LLM wrapper for Palantir-provided language models."""
def __init__(self, model: OpenAiGptChatLanguageModel):
self.model = model
logger.info("PalantirLLM initialized with model: %s", model)
def invoke(
self,
prompt: str,
stop: Optional[List[str]] = None,
**kwargs: Any,
) -> str:
"""Run the LLM on the given input."""
logger.info("Generating response for prompt: %s", prompt)
system_prompt = "Answer the following question"
request = GptChatCompletionRequest(
[ChatMessage(ChatMessageRole.SYSTEM, system_prompt), ChatMessage(ChatMessageRole.USER, prompt)]
)
try:
resp = self.model.create_chat_completion(request)
logger.info("Response received: %s", resp.choices[0].message.content)
return resp.choices[0].message.content
except Exception as e:
logger.error("Error during API call: %s", e)
return "Error generating response"
@property
def _llm_type(self) -> str:
"""Get the type of language model used by this chat model. Used for logging purposes only."""
return "palantir"
@transform(
model=OpenAiGptChatLanguageModelInput("ri.language-model-service..language-model.gpt-4-o"),
output=Output("ri.foundry.main.dataset.83708b87-7551-44f3-8579-bce00edb67e5"),
)
def compute_answer(ctx, model: OpenAiGptChatLanguageModel, output):
logger.info("Starting compute_answer transform")
question = "ChatGPT에 대해 설명 해 주세요"
palantir_llm = PalantirLLM(model)
answer = palantir_llm.invoke(question)
logger.info("Generated answer: %s", answer)
# Assuming you want to save the answer to a dataset
try:
answer_df = pd.DataFrame({"question": [question], "answer": [answer]})
spark_df = ctx.spark_session.createDataFrame(answer_df)
output.write_dataframe(spark_df)
logger.info("Answer saved to dataset")
except Exception as e:
logger.error("Error saving to dataset: %s", e)
Are you getting an error? You could try using the debugger to inspect it and make sure it’s behaving like you expect.
There is no error but I cannot use full Langchain option with that wrapping code…
Hi Jacob,
My team was also interested in getting Langchain to work with the palantir_models (https://www.palantir.com/docs/foundry/integrate-models/language-models-transforms-inputs#openaigptchatlanguagemodelinput) library. Through looking at your code and Langchain custom LLM class documentation (https://python.langchain.com/docs/how_to/custom_llm/), here is how I wrapped the Palantir model to get it to run in a 1) Jupyter code repository and 2) Python transforms repository. You’ll have to modify the code slightly for your use case. Hope this helps!
1) Jupyter code repository
from typing import Any, Dict, Iterator, List, Mapping, Optional
from langchain_core.callbacks.manager import CallbackManagerForLLMRun
from langchain_core.language_models.llms import LLM
from langchain_core.outputs import GenerationChunk
class CustomLLM(LLM):
"""A custom chat model that echoes the first `n` characters of the input.
When contributing an implementation to LangChain, carefully document
the model including the initialization parameters, include
an example of how to initialize the model and include any relevant
links to the underlying models documentation or API.
Example:
.. code-block:: python
model = CustomChatModel(n=2)
result = model.invoke([HumanMessage(content="hello")])
result = model.batch([[HumanMessage(content="hello")],
[HumanMessage(content="world")]])
"""
model: OpenAiGptChatLanguageModel
"""The number of characters from the last message of the prompt to be echoed."""
def _call(
self,
prompt: str,
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> str:
"""Run the LLM on the given input.
Override this method to implement the LLM logic.
Args:
prompt: The prompt to generate from.
stop: Stop words to use when generating. Model output is cut off at the
first occurrence of any of the stop substrings.
If stop tokens are not supported consider raising NotImplementedError.
run_manager: Callback manager for the run.
**kwargs: Arbitrary additional keyword arguments. These are usually passed
to the model provider API call.
Returns:
The model output as a string. Actual completions SHOULD NOT include the prompt.
"""
system_prompt = "Answer the following question"
request = GptChatCompletionRequest(
[ChatMessage(ChatMessageRole.SYSTEM, system_prompt), ChatMessage(ChatMessageRole.USER, prompt)]
)
try:
resp = self.model.create_chat_completion(request)
return resp.choices[0].message.content
except Exception as e:
return "Error generating response"
def _stream(
self,
prompt: str,
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> Iterator[GenerationChunk]:
"""Stream the LLM on the given prompt.
This method should be overridden by subclasses that support streaming.
If not implemented, the default behavior of calls to stream will be to
fallback to the non-streaming version of the model and return
the output as a single chunk.
Args:
prompt: The prompt to generate from.
stop: Stop words to use when generating. Model output is cut off at the
first occurrence of any of these substrings.
run_manager: Callback manager for the run.
**kwargs: Arbitrary additional keyword arguments. These are usually passed
to the model provider API call.
Returns:
An iterator of GenerationChunks.
"""
for char in prompt[: self.n]:
chunk = GenerationChunk(text=char)
if run_manager:
run_manager.on_llm_new_token(chunk.text, chunk=chunk)
yield chunk
@property
def _identifying_params(self) -> Dict[str, Any]:
"""Return a dictionary of identifying parameters."""
return {
# The model name allows users to specify custom token counting
# rules in LLM monitoring applications (e.g., in LangSmith users
# can provide per token pricing for their model and monitor
# costs for the given LLM.)
"model_name": "CustomChatModel",
}
@property
def _llm_type(self) -> str:
"""Get the type of language model used by this chat model. Used for logging purposes only."""
return "custom"
model = OpenAiGptChatLanguageModel.get("GPT_4o")
llm = CustomLLM(model=model)
print(llm)
from langchain_core.prompts import PromptTemplate
llm.invoke("How are you doing?")
2. Python transforms repository
from typing import Any, Dict, Iterator, List, Mapping, Optional
from langchain_core.callbacks.manager import CallbackManagerForLLMRun
from langchain_core.language_models.llms import LLM
from langchain_core.outputs import GenerationChunk
from pyspark.sql import DataFrame
from transforms.api import Input, Output, transform
from palantir_models.transforms import OpenAiGptChatLanguageModelInput
from palantir_models.models import OpenAiGptChatLanguageModel
from language_model_service_api.languagemodelservice_api_completion_v3 import GptChatCompletionRequest
from language_model_service_api.languagemodelservice_api import ChatMessage, ChatMessageRole
import pandas as pd
from myproject.datasets import utils
class CustomLLM(LLM):
"""A custom chat model that echoes the first `n` characters of the input.
When contributing an implementation to LangChain, carefully document
the model including the initialization parameters, include
an example of how to initialize the model and include any relevant
links to the underlying models documentation or API.
Example:
.. code-block:: python
model = CustomChatModel(n=2)
result = model.invoke([HumanMessage(content="hello")])
result = model.batch([[HumanMessage(content="hello")],
[HumanMessage(content="world")]])
"""
model: OpenAiGptChatLanguageModel
"""The number of characters from the last message of the prompt to be echoed."""
def _call(
self,
prompt: str,
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> str:
"""Run the LLM on the given input.
Override this method to implement the LLM logic.
Args:
prompt: The prompt to generate from.
stop: Stop words to use when generating. Model output is cut off at the
first occurrence of any of the stop substrings.
If stop tokens are not supported consider raising NotImplementedError.
run_manager: Callback manager for the run.
**kwargs: Arbitrary additional keyword arguments. These are usually passed
to the model provider API call.
Returns:
The model output as a string. Actual completions SHOULD NOT include the prompt.
"""
system_prompt = "Answer the following question"
request = GptChatCompletionRequest(
[ChatMessage(ChatMessageRole.SYSTEM, system_prompt), ChatMessage(ChatMessageRole.USER, prompt)]
)
try:
resp = self.model.create_chat_completion(request)
return resp.choices[0].message.content
except Exception as e:
print("Error during API call: %s", e)
return "Error generating response"
def _stream(
self,
prompt: str,
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> Iterator[GenerationChunk]:
"""Stream the LLM on the given prompt.
This method should be overridden by subclasses that support streaming.
If not implemented, the default behavior of calls to stream will be to
fallback to the non-streaming version of the model and return
the output as a single chunk.
Args:
prompt: The prompt to generate from.
stop: Stop words to use when generating. Model output is cut off at the
first occurrence of any of these substrings.
run_manager: Callback manager for the run.
**kwargs: Arbitrary additional keyword arguments. These are usually passed
to the model provider API call.
Returns:
An iterator of GenerationChunks.
"""
for char in prompt[: self.n]:
chunk = GenerationChunk(text=char)
if run_manager:
run_manager.on_llm_new_token(chunk.text, chunk=chunk)
yield chunk
@property
def _identifying_params(self) -> Dict[str, Any]:
"""Return a dictionary of identifying parameters."""
return {
# The model name allows users to specify custom token counting
# rules in LLM monitoring applications (e.g., in LangSmith users
# can provide per token pricing for their model and monitor
# costs for the given LLM.)
"model_name": "CustomChatModel",
}
@property
def _llm_type(self) -> str:
"""Get the type of language model used by this chat model. Used for logging purposes only."""
return "custom"
@transform(
model=OpenAiGptChatLanguageModelInput("ri.language-model-service..language-model.gpt-4-o"),
output=Output("insert-your-output-rid-here")
)
def compute_answer(ctx, model: OpenAiGptChatLanguageModel, output):
import numpy as np
question = "Why is the sky blue?"
wrapped_llm = CustomLLM(model=model)
answer = wrapped_llm.invoke(question)
try:
answer_df = pd.DataFrame({"question": [question], "answer": [answer]})
spark_df = ctx.spark_session.createDataFrame(answer_df)
output.write_dataframe(spark_df)
except Exception as e:
print(f"Error saving to dataset: {e}")
1 Like
This topic was automatically closed 14 days after the last reply. New replies are no longer allowed.