mirror of
https://github.com/langchain-ai/langchain.git
synced 2026-10-05 01:15:09 +03:00
chore(infra): delete prompty (#35044)
This commit is contained in:
ccurme authored and GitHub committed
1 parent
81b4752419
commit
f058e45dfb
39 files changed
+1
-3813
No files matched your search
@@ -66,7 +66,6 @@ body:
|
||||
- label: langchain-nomic
|
||||
- label: langchain-ollama
|
||||
- label: langchain-perplexity
|
||||
- label: langchain-prompty
|
||||
- label: langchain-qdrant
|
||||
- label: langchain-xai
|
||||
- label: Other / not sure / general
|
||||
|
||||
@@ -63,7 +63,6 @@ body:
|
||||
- label: langchain-nomic
|
||||
- label: langchain-ollama
|
||||
- label: langchain-perplexity
|
||||
- label: langchain-prompty
|
||||
- label: langchain-qdrant
|
||||
- label: langchain-xai
|
||||
- label: Other / not sure / general
|
||||
|
||||
@@ -43,7 +43,6 @@ body:
|
||||
- label: langchain-nomic
|
||||
- label: langchain-ollama
|
||||
- label: langchain-perplexity
|
||||
- label: langchain-prompty
|
||||
- label: langchain-qdrant
|
||||
- label: langchain-xai
|
||||
- label: Other / not sure / general
|
||||
@@ -114,7 +114,6 @@ body:
|
||||
- label: langchain-nomic
|
||||
- label: langchain-ollama
|
||||
- label: langchain-perplexity
|
||||
- label: langchain-prompty
|
||||
- label: langchain-qdrant
|
||||
- label: langchain-xai
|
||||
- label: Other / not sure / general
|
||||
@@ -98,11 +98,6 @@ perplexity:
|
||||
- any-glob-to-any-file:
|
||||
- "libs/partners/perplexity/**/*"
|
||||
|
||||
prompty:
|
||||
- changed-files:
|
||||
- any-glob-to-any-file:
|
||||
- "libs/partners/prompty/**/*"
|
||||
|
||||
qdrant:
|
||||
- changed-files:
|
||||
- any-glob-to-any-file:
|
||||
|
||||
@@ -46,9 +46,6 @@ IGNORED_PARTNERS = [
|
||||
# specifically in huggingface jobs
|
||||
# https://github.com/langchain-ai/langchain/issues/25558
|
||||
"huggingface",
|
||||
# prompty exhibiting issues with numpy for Python 3.13
|
||||
# https://github.com/langchain-ai/langchain/actions/runs/12651104685/job/35251034969?pr=29065
|
||||
"prompty",
|
||||
]
|
||||
|
||||
|
||||
|
||||
@@ -43,7 +43,6 @@ jobs:
|
||||
"langchain-nomic": "nomic",
|
||||
"langchain-ollama": "ollama",
|
||||
"langchain-perplexity": "perplexity",
|
||||
"langchain-prompty": "prompty",
|
||||
"langchain-qdrant": "qdrant",
|
||||
"langchain-xai": "xai",
|
||||
};
|
||||
|
||||
@@ -30,7 +30,7 @@
|
||||
# core, langchain, langchain-classic, model-profiles,
|
||||
# standard-tests, text-splitters, docs, anthropic, chroma, deepseek, exa,
|
||||
# fireworks, groq, huggingface, mistralai, nomic, ollama, openai,
|
||||
# perplexity, prompty, qdrant, xai, infra, deps
|
||||
# perplexity, qdrant, xai, infra, deps
|
||||
#
|
||||
# Multiple scopes can be used by separating them with a comma. For example:
|
||||
#
|
||||
@@ -103,7 +103,6 @@ jobs:
|
||||
ollama
|
||||
openai
|
||||
perplexity
|
||||
prompty
|
||||
qdrant
|
||||
xai
|
||||
infra
|
||||
|
||||
@@ -105,12 +105,6 @@ repos:
|
||||
entry: make -C libs/partners/openai format lint
|
||||
files: ^libs/partners/openai/
|
||||
pass_filenames: false
|
||||
- id: prompty
|
||||
name: format and lint partners/prompty
|
||||
language: system
|
||||
entry: make -C libs/partners/prompty format lint
|
||||
files: ^libs/partners/prompty/
|
||||
pass_filenames: false
|
||||
- id: qdrant
|
||||
name: format and lint partners/qdrant
|
||||
language: system
|
||||
|
||||
@@ -1 +0,0 @@
|
||||
__pycache__
|
||||
@@ -1,21 +0,0 @@
|
||||
MIT License
|
||||
|
||||
Copyright (c) 2023 LangChain, Inc.
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
in the Software without restriction, including without limitation the rights
|
||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
copies of the Software, and to permit persons to whom the Software is
|
||||
furnished to do so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
SOFTWARE.
|
||||
@@ -1,62 +0,0 @@
|
||||
.PHONY: all format lint type test tests integration_tests help extended_tests
|
||||
|
||||
# Default target executed when no arguments are given to make.
|
||||
all: help
|
||||
|
||||
.EXPORT_ALL_VARIABLES:
|
||||
UV_FROZEN = true
|
||||
|
||||
# Define a variable for the test file path.
|
||||
TEST_FILE ?= tests/unit_tests/
|
||||
|
||||
test:
|
||||
uv run --group test pytest $(TEST_FILE)
|
||||
|
||||
tests:
|
||||
uv run --group test pytest $(TEST_FILE)
|
||||
|
||||
test_watch:
|
||||
uv run --group test ptw --snapshot-update --now . -- -vv $(TEST_FILE)
|
||||
|
||||
|
||||
######################
|
||||
# LINTING AND FORMATTING
|
||||
######################
|
||||
|
||||
# Define a variable for Python and notebook files.
|
||||
PYTHON_FILES=.
|
||||
MYPY_CACHE=.mypy_cache
|
||||
lint format: PYTHON_FILES=.
|
||||
lint_diff format_diff: PYTHON_FILES=$(shell git diff --relative=libs/partners/prompty --name-only --diff-filter=d master | grep -E '\.py$$|\.ipynb$$')
|
||||
lint_package: PYTHON_FILES=langchain_prompty
|
||||
lint_tests: PYTHON_FILES=tests
|
||||
lint_tests: MYPY_CACHE=.mypy_cache_test
|
||||
|
||||
lint lint_diff lint_package lint_tests:
|
||||
[ "$(PYTHON_FILES)" = "" ] || uv run --all-groups ruff check $(PYTHON_FILES)
|
||||
[ "$(PYTHON_FILES)" = "" ] || uv run --all-groups ruff format $(PYTHON_FILES) --diff
|
||||
[ "$(PYTHON_FILES)" = "" ] || mkdir -p $(MYPY_CACHE) && uv run --all-groups mypy $(PYTHON_FILES) --cache-dir $(MYPY_CACHE)
|
||||
|
||||
type:
|
||||
mkdir -p $(MYPY_CACHE) && uv run --all-groups mypy $(PYTHON_FILES) --cache-dir $(MYPY_CACHE)
|
||||
|
||||
format format_diff:
|
||||
[ "$(PYTHON_FILES)" = "" ] || uv run --all-groups ruff format $(PYTHON_FILES)
|
||||
[ "$(PYTHON_FILES)" = "" ] || uv run --all-groups ruff check --fix $(PYTHON_FILES)
|
||||
|
||||
check_imports: $(shell find langchain_prompty -name '*.py')
|
||||
uv run --all-groups python ./scripts/check_imports.py $^
|
||||
|
||||
######################
|
||||
# HELP
|
||||
######################
|
||||
|
||||
help:
|
||||
@echo '----'
|
||||
@echo 'check_imports - check imports'
|
||||
@echo 'format - run code formatters'
|
||||
@echo 'lint - run linters'
|
||||
@echo 'type - run type checking'
|
||||
@echo 'test - run unit tests'
|
||||
@echo 'tests - run unit tests'
|
||||
@echo 'test TEST_FILE=<test_file> - run all tests in file'
|
||||
@@ -1,69 +0,0 @@
|
||||
# langchain-prompty
|
||||
|
||||
[](https://pypi.org/project/langchain-prompty/#history)
|
||||
[](https://opensource.org/licenses/MIT)
|
||||
[](https://pypistats.org/packages/langchain-prompty)
|
||||
[](https://x.com/langchain)
|
||||
|
||||
Looking for the JS/TS version? Check out [LangChain.js](https://github.com/langchain-ai/langchainjs).
|
||||
|
||||
## Quick Install
|
||||
|
||||
```bash
|
||||
pip install langchain-prompty
|
||||
```
|
||||
|
||||
## 🤔 What is this?
|
||||
|
||||
This package contains the LangChain integration with Microsoft Prompty.
|
||||
|
||||
## 📖 Documentation
|
||||
|
||||
View the [documentation](https://docs.langchain.com/oss/python/integrations/providers/microsoft) for more details.
|
||||
|
||||
## Usage
|
||||
|
||||
Use the `create_chat_prompt` function to load `prompty` file as prompt.
|
||||
|
||||
```python
|
||||
from langchain_prompty import create_chat_prompt
|
||||
|
||||
prompt = create_chat_prompt('<your .prompty file path>')
|
||||
```
|
||||
|
||||
Then you can use the prompt for next steps.
|
||||
|
||||
Here is an example .prompty file:
|
||||
|
||||
```prompty
|
||||
---
|
||||
name: Basic Prompt
|
||||
description: A basic prompt that uses the GPT-3 chat API to answer questions
|
||||
authors:
|
||||
- author_1
|
||||
- author_2
|
||||
model:
|
||||
api: chat
|
||||
configuration:
|
||||
azure_deployment: gpt-35-turbo
|
||||
sample:
|
||||
firstName: Jane
|
||||
lastName: Doe
|
||||
question: What is the meaning of life?
|
||||
chat_history: []
|
||||
---
|
||||
system:
|
||||
You are an AI assistant who helps people find information.
|
||||
As the assistant, you answer questions briefly, succinctly,
|
||||
and in a personable manner using markdown and even add some personal flair with appropriate emojis.
|
||||
|
||||
{% for item in chat_history %}
|
||||
{{item.role}}:
|
||||
{{item.content}}
|
||||
{% endfor %}
|
||||
|
||||
|
||||
user:
|
||||
{{input}}
|
||||
|
||||
```
|
||||
@@ -1,11 +0,0 @@
|
||||
"""Microsoft Prompty integration for LangChain."""
|
||||
|
||||
from langchain_prompty.core import InvokerFactory
|
||||
from langchain_prompty.langchain import create_chat_prompt
|
||||
from langchain_prompty.parsers import PromptyChatParser
|
||||
from langchain_prompty.renderers import MustacheRenderer
|
||||
|
||||
InvokerFactory().register_renderer("mustache", MustacheRenderer)
|
||||
InvokerFactory().register_parser("prompty.chat", PromptyChatParser)
|
||||
|
||||
__all__ = ["create_chat_prompt"]
|
||||
@@ -1,336 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import abc
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from typing import Any, Generic, Literal, TypeVar
|
||||
|
||||
import yaml
|
||||
from pydantic import BaseModel, ConfigDict, Field, FilePath
|
||||
|
||||
T = TypeVar("T")
|
||||
|
||||
|
||||
class SimpleModel(BaseModel, Generic[T]):
|
||||
"""Simple model for a single item."""
|
||||
|
||||
item: T
|
||||
|
||||
|
||||
class PropertySettings(BaseModel):
|
||||
"""Property settings for a prompty model."""
|
||||
|
||||
model_config = ConfigDict(arbitrary_types_allowed=True)
|
||||
type: Literal["string", "number", "array", "object", "boolean"]
|
||||
default: str | int | float | list | dict | bool | None = Field(default=None)
|
||||
description: str = Field(default="")
|
||||
|
||||
|
||||
class ModelSettings(BaseModel):
|
||||
"""Model settings for a prompty model."""
|
||||
|
||||
api: str = Field(default="")
|
||||
configuration: dict = Field(default={})
|
||||
parameters: dict = Field(default={})
|
||||
response: dict = Field(default={})
|
||||
|
||||
def model_dump_safe(self) -> dict:
|
||||
d = self.model_dump()
|
||||
d["configuration"] = {
|
||||
k: "*" * len(v) if "key" in k.lower() or "secret" in k.lower() else v
|
||||
for k, v in d["configuration"].items()
|
||||
}
|
||||
return d
|
||||
|
||||
|
||||
class TemplateSettings(BaseModel):
|
||||
"""Template settings for a prompty model."""
|
||||
|
||||
type: str = Field(default="mustache")
|
||||
parser: str = Field(default="")
|
||||
|
||||
|
||||
class Prompty(BaseModel):
|
||||
"""Base Prompty model."""
|
||||
|
||||
# Metadata
|
||||
name: str = Field(default="")
|
||||
description: str = Field(default="")
|
||||
authors: list[str] = Field(default=[])
|
||||
tags: list[str] = Field(default=[])
|
||||
version: str = Field(default="")
|
||||
base: str = Field(default="")
|
||||
basePrompty: Prompty | None = Field(default=None)
|
||||
|
||||
# Model
|
||||
model: ModelSettings = Field(default_factory=ModelSettings)
|
||||
|
||||
# Sample
|
||||
sample: dict = Field(default={})
|
||||
|
||||
# Input / output
|
||||
inputs: dict[str, PropertySettings] = Field(default={})
|
||||
outputs: dict[str, PropertySettings] = Field(default={})
|
||||
|
||||
# Template
|
||||
template: TemplateSettings
|
||||
|
||||
file: FilePath = Field(default="") # type: ignore[assignment]
|
||||
content: str = Field(default="")
|
||||
|
||||
def to_safe_dict(self) -> dict[str, Any]:
|
||||
d = {}
|
||||
for k, v in self:
|
||||
if v != "" and v != {} and v != [] and v is not None:
|
||||
if k == "model":
|
||||
d[k] = v.model_dump_safe()
|
||||
elif k == "template":
|
||||
d[k] = v.model_dump()
|
||||
elif k == "inputs" or k == "outputs":
|
||||
d[k] = {k: v.model_dump() for k, v in v.items()}
|
||||
elif k == "file":
|
||||
d[k] = (
|
||||
str(self.file.as_posix())
|
||||
if isinstance(self.file, Path)
|
||||
else self.file
|
||||
)
|
||||
elif k == "basePrompty":
|
||||
# No need to serialize basePrompty
|
||||
continue
|
||||
|
||||
else:
|
||||
d[k] = v
|
||||
return d
|
||||
|
||||
# Generate json representation of the prompty
|
||||
def to_safe_json(self) -> str:
|
||||
d = self.to_safe_dict()
|
||||
return json.dumps(d)
|
||||
|
||||
@staticmethod
|
||||
def normalize(attribute: Any, parent: Path, env_error: bool = True) -> Any:
|
||||
if isinstance(attribute, str):
|
||||
attribute = attribute.strip()
|
||||
if attribute.startswith("${") and attribute.endswith("}"):
|
||||
variable = attribute[2:-1].split(":")
|
||||
if variable[0] in os.environ.keys():
|
||||
return os.environ[variable[0]]
|
||||
else:
|
||||
if len(variable) > 1:
|
||||
return variable[1]
|
||||
else:
|
||||
if env_error:
|
||||
raise ValueError(
|
||||
f"Variable {variable[0]} not found in environment"
|
||||
)
|
||||
else:
|
||||
return ""
|
||||
elif (
|
||||
attribute.startswith("file:")
|
||||
and Path(parent / attribute.split(":")[1]).exists()
|
||||
):
|
||||
with open(parent / attribute.split(":")[1]) as f:
|
||||
items = json.load(f)
|
||||
if isinstance(items, list):
|
||||
return [Prompty.normalize(value, parent) for value in items]
|
||||
elif isinstance(items, dict):
|
||||
return {
|
||||
key: Prompty.normalize(value, parent)
|
||||
for key, value in items.items()
|
||||
}
|
||||
else:
|
||||
return items
|
||||
else:
|
||||
return attribute
|
||||
elif isinstance(attribute, list):
|
||||
return [Prompty.normalize(value, parent) for value in attribute]
|
||||
elif isinstance(attribute, dict):
|
||||
return {
|
||||
key: Prompty.normalize(value, parent)
|
||||
for key, value in attribute.items()
|
||||
}
|
||||
else:
|
||||
return attribute
|
||||
|
||||
|
||||
def param_hoisting(
|
||||
top: dict[str, Any], bottom: dict[str, Any], top_key: Any = None
|
||||
) -> dict[str, Any]:
|
||||
"""Merge two dictionaries with hoisting of parameters from bottom to top.
|
||||
|
||||
Args:
|
||||
top: The top dictionary.
|
||||
bottom: The bottom dictionary.
|
||||
top_key: The key to hoist from the bottom to the top.
|
||||
|
||||
Returns:
|
||||
The merged dictionary.
|
||||
"""
|
||||
if top_key:
|
||||
new_dict = {**top[top_key]} if top_key in top else {}
|
||||
else:
|
||||
new_dict = {**top}
|
||||
for key, value in bottom.items():
|
||||
if key not in new_dict:
|
||||
new_dict[key] = value
|
||||
return new_dict
|
||||
|
||||
|
||||
class Invoker(abc.ABC):
|
||||
"""Base class for all invokers."""
|
||||
|
||||
def __init__(self, prompty: Prompty) -> None:
|
||||
self.prompty = prompty
|
||||
|
||||
@abc.abstractmethod
|
||||
def invoke(self, data: BaseModel) -> BaseModel:
|
||||
pass
|
||||
|
||||
def __call__(self, data: BaseModel) -> BaseModel:
|
||||
return self.invoke(data)
|
||||
|
||||
|
||||
class NoOpParser(Invoker):
|
||||
"""NoOp parser for invokers."""
|
||||
|
||||
def invoke(self, data: BaseModel) -> BaseModel:
|
||||
return data
|
||||
|
||||
|
||||
class InvokerFactory:
|
||||
"""Factory for creating invokers.
|
||||
|
||||
This class implements a thread-safe singleton pattern using double-checked
|
||||
locking to ensure only one instance is created even under concurrent access.
|
||||
"""
|
||||
|
||||
_instance = None
|
||||
_lock = threading.Lock()
|
||||
_renderers: dict[str, type[Invoker]] = {}
|
||||
_parsers: dict[str, type[Invoker]] = {}
|
||||
_executors: dict[str, type[Invoker]] = {}
|
||||
_processors: dict[str, type[Invoker]] = {}
|
||||
|
||||
def __new__(cls) -> InvokerFactory:
|
||||
# First check (fast path, no lock needed)
|
||||
if cls._instance is None:
|
||||
with cls._lock:
|
||||
# Double-check after acquiring lock
|
||||
if cls._instance is None:
|
||||
cls._instance = super().__new__(cls)
|
||||
# Add NOOP invokers
|
||||
cls._renderers["NOOP"] = NoOpParser
|
||||
cls._parsers["NOOP"] = NoOpParser
|
||||
cls._executors["NOOP"] = NoOpParser
|
||||
cls._processors["NOOP"] = NoOpParser
|
||||
return cls._instance
|
||||
|
||||
def register(
|
||||
self,
|
||||
type: Literal["renderer", "parser", "executor", "processor"],
|
||||
name: str,
|
||||
invoker: type[Invoker],
|
||||
) -> None:
|
||||
if type == "renderer":
|
||||
self._renderers[name] = invoker
|
||||
elif type == "parser":
|
||||
self._parsers[name] = invoker
|
||||
elif type == "executor":
|
||||
self._executors[name] = invoker
|
||||
elif type == "processor":
|
||||
self._processors[name] = invoker
|
||||
else:
|
||||
raise ValueError(f"Invalid type {type}")
|
||||
|
||||
def register_renderer(self, name: str, renderer_class: Any) -> None:
|
||||
self.register("renderer", name, renderer_class)
|
||||
|
||||
def register_parser(self, name: str, parser_class: Any) -> None:
|
||||
self.register("parser", name, parser_class)
|
||||
|
||||
def register_executor(self, name: str, executor_class: Any) -> None:
|
||||
self.register("executor", name, executor_class)
|
||||
|
||||
def register_processor(self, name: str, processor_class: Any) -> None:
|
||||
self.register("processor", name, processor_class)
|
||||
|
||||
def __call__(
|
||||
self,
|
||||
type: Literal["renderer", "parser", "executor", "processor"],
|
||||
name: str,
|
||||
prompty: Prompty,
|
||||
data: BaseModel,
|
||||
) -> Any:
|
||||
if type == "renderer":
|
||||
return self._renderers[name](prompty)(data)
|
||||
elif type == "parser":
|
||||
return self._parsers[name](prompty)(data)
|
||||
elif type == "executor":
|
||||
return self._executors[name](prompty)(data)
|
||||
elif type == "processor":
|
||||
return self._processors[name](prompty)(data)
|
||||
else:
|
||||
raise ValueError(f"Invalid type {type}")
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"renderers": {
|
||||
k: f"{v.__module__}.{v.__name__}" for k, v in self._renderers.items()
|
||||
},
|
||||
"parsers": {
|
||||
k: f"{v.__module__}.{v.__name__}" for k, v in self._parsers.items()
|
||||
},
|
||||
"executors": {
|
||||
k: f"{v.__module__}.{v.__name__}" for k, v in self._executors.items()
|
||||
},
|
||||
"processors": {
|
||||
k: f"{v.__module__}.{v.__name__}" for k, v in self._processors.items()
|
||||
},
|
||||
}
|
||||
|
||||
def to_json(self) -> str:
|
||||
return json.dumps(self.to_dict())
|
||||
|
||||
|
||||
class Frontmatter:
|
||||
"""Class for reading frontmatter from a string or file."""
|
||||
|
||||
_yaml_delim = r"(?:---|\+\+\+)"
|
||||
_yaml = r"(.*?)"
|
||||
_content = r"\s*(.+)$"
|
||||
_re_pattern = r"^\s*" + _yaml_delim + _yaml + _yaml_delim + _content
|
||||
_regex = re.compile(_re_pattern, re.S | re.M)
|
||||
|
||||
@classmethod
|
||||
def read_file(cls, path: str) -> dict[str, Any]:
|
||||
"""Reads file at path and returns dict with separated frontmatter.
|
||||
See read() for more info on dict return value.
|
||||
"""
|
||||
with open(path, encoding="utf-8") as file:
|
||||
file_contents = file.read()
|
||||
return cls.read(file_contents)
|
||||
|
||||
@classmethod
|
||||
def read(cls, string: str) -> dict[str, Any]:
|
||||
"""Returns dict with separated frontmatter from string.
|
||||
|
||||
Returned dict keys:
|
||||
- attributes: extracted YAML attributes in dict form.
|
||||
- body: string contents below the YAML separators
|
||||
- frontmatter: string representation of YAML
|
||||
"""
|
||||
fmatter = ""
|
||||
body = ""
|
||||
result = cls._regex.search(string)
|
||||
|
||||
if result:
|
||||
fmatter = result.group(1)
|
||||
body = result.group(2)
|
||||
return {
|
||||
"attributes": yaml.safe_load(fmatter),
|
||||
"body": body,
|
||||
"frontmatter": fmatter,
|
||||
}
|
||||
@@ -1,35 +0,0 @@
|
||||
from typing import Any
|
||||
|
||||
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
|
||||
from langchain_core.runnables import Runnable, RunnableLambda
|
||||
|
||||
from .parsers import RoleMap
|
||||
from .utils import load, prepare
|
||||
|
||||
|
||||
def create_chat_prompt(
|
||||
path: str,
|
||||
input_name_agent_scratchpad: str = "agent_scratchpad",
|
||||
) -> Runnable[dict[str, Any], ChatPromptTemplate]:
|
||||
"""Create a chat prompt from a LangChain schema."""
|
||||
|
||||
def runnable_chat_lambda(inputs: dict[str, Any]) -> ChatPromptTemplate:
|
||||
p = load(path)
|
||||
parsed = prepare(p, inputs)
|
||||
# Parsed messages have been templated
|
||||
# Convert to Message objects to avoid templating attempts in ChatPromptTemplate
|
||||
lc_messages = []
|
||||
for message in parsed:
|
||||
message_class = RoleMap.get_message_class(message["role"])
|
||||
lc_messages.append(message_class(content=message["content"]))
|
||||
|
||||
lc_messages.append(
|
||||
MessagesPlaceholder(
|
||||
variable_name=input_name_agent_scratchpad, optional=True
|
||||
) # type: ignore[arg-type]
|
||||
)
|
||||
lc_p = ChatPromptTemplate.from_messages(lc_messages)
|
||||
lc_p = lc_p.partial(**p.inputs)
|
||||
return lc_p
|
||||
|
||||
return RunnableLambda(runnable_chat_lambda)
|
||||
@@ -1,133 +0,0 @@
|
||||
import base64
|
||||
import re
|
||||
|
||||
from langchain_core.messages import (
|
||||
AIMessage,
|
||||
BaseMessage,
|
||||
FunctionMessage,
|
||||
HumanMessage,
|
||||
SystemMessage,
|
||||
)
|
||||
from pydantic import BaseModel
|
||||
|
||||
from .core import Invoker, Prompty, SimpleModel
|
||||
|
||||
|
||||
class RoleMap:
|
||||
_ROLE_MAP: dict[str, type[BaseMessage]] = {
|
||||
"system": SystemMessage,
|
||||
"user": HumanMessage,
|
||||
"human": HumanMessage,
|
||||
"assistant": AIMessage,
|
||||
"ai": AIMessage,
|
||||
"function": FunctionMessage,
|
||||
}
|
||||
ROLES = _ROLE_MAP.keys()
|
||||
|
||||
@classmethod
|
||||
def get_message_class(cls, role: str) -> type[BaseMessage]:
|
||||
return cls._ROLE_MAP[role]
|
||||
|
||||
|
||||
class PromptyChatParser(Invoker):
|
||||
"""Parse a chat prompt into a list of messages."""
|
||||
|
||||
def __init__(self, prompty: Prompty) -> None:
|
||||
self.prompty = prompty
|
||||
self.roles = RoleMap.ROLES
|
||||
self.path = self.prompty.file.parent
|
||||
|
||||
def inline_image(self, image_item: str) -> str:
|
||||
# pass through if it's a url or base64 encoded
|
||||
if image_item.startswith("http") or image_item.startswith("data"):
|
||||
return image_item
|
||||
# otherwise, it's a local file - need to base64 encode it
|
||||
else:
|
||||
image_path = self.path / image_item
|
||||
with open(image_path, "rb") as f:
|
||||
base64_image = base64.b64encode(f.read()).decode("utf-8")
|
||||
|
||||
if image_path.suffix == ".png":
|
||||
return f"data:image/png;base64,{base64_image}"
|
||||
elif image_path.suffix == ".jpg":
|
||||
return f"data:image/jpeg;base64,{base64_image}"
|
||||
elif image_path.suffix == ".jpeg":
|
||||
return f"data:image/jpeg;base64,{base64_image}"
|
||||
else:
|
||||
raise ValueError(
|
||||
f"Invalid image format {image_path.suffix} - currently only .png "
|
||||
"and .jpg / .jpeg are supported."
|
||||
)
|
||||
|
||||
def parse_content(self, content: str) -> str | list:
|
||||
"""for parsing inline images"""
|
||||
# regular expression to parse markdown images
|
||||
image = r"(?P<alt>!\[[^\]]*\])\((?P<filename>.*?)(?=\"|\))\)"
|
||||
matches = re.findall(image, content, flags=re.MULTILINE)
|
||||
if len(matches) > 0:
|
||||
content_items = []
|
||||
content_chunks = re.split(image, content, flags=re.MULTILINE)
|
||||
current_chunk = 0
|
||||
for i in range(len(content_chunks)):
|
||||
# image entry
|
||||
if (
|
||||
current_chunk < len(matches)
|
||||
and content_chunks[i] == matches[current_chunk][0]
|
||||
):
|
||||
content_items.append(
|
||||
{
|
||||
"type": "image_url",
|
||||
"image_url": {
|
||||
"url": self.inline_image(
|
||||
matches[current_chunk][1].split(" ")[0].strip()
|
||||
)
|
||||
},
|
||||
}
|
||||
)
|
||||
# second part of image entry
|
||||
elif (
|
||||
current_chunk < len(matches)
|
||||
and content_chunks[i] == matches[current_chunk][1]
|
||||
):
|
||||
current_chunk += 1
|
||||
# text entry
|
||||
else:
|
||||
if len(content_chunks[i].strip()) > 0:
|
||||
content_items.append(
|
||||
{"type": "text", "text": content_chunks[i].strip()}
|
||||
)
|
||||
return content_items
|
||||
else:
|
||||
return content
|
||||
|
||||
def invoke(self, data: BaseModel) -> BaseModel:
|
||||
if not isinstance(data, SimpleModel):
|
||||
raise ValueError("data must be an instance of SimpleModel")
|
||||
messages = []
|
||||
separator = r"(?i)^\s*#?\s*(" + "|".join(self.roles) + r")\s*:\s*\n"
|
||||
|
||||
# get valid chunks - remove empty items
|
||||
chunks = [
|
||||
item
|
||||
for item in re.split(separator, data.item, flags=re.MULTILINE)
|
||||
if len(item.strip()) > 0
|
||||
]
|
||||
|
||||
# if no starter role, then inject system role
|
||||
if chunks[0].strip().lower() not in self.roles:
|
||||
chunks.insert(0, "system")
|
||||
|
||||
# if last chunk is role entry, then remove (no content?)
|
||||
if chunks[-1].strip().lower() in self.roles:
|
||||
chunks.pop()
|
||||
|
||||
if len(chunks) % 2 != 0:
|
||||
raise ValueError("Invalid prompt format")
|
||||
|
||||
# create messages
|
||||
for i in range(0, len(chunks), 2):
|
||||
role = chunks[i].strip().lower()
|
||||
content = chunks[i + 1].strip()
|
||||
messages.append({"role": role, "content": self.parse_content(content)})
|
||||
|
||||
return SimpleModel[list](item=messages)
|
||||
@@ -1,39 +0,0 @@
|
||||
from typing import Any
|
||||
|
||||
from langchain_core.messages import BaseMessage
|
||||
from langchain_core.utils import mustache
|
||||
from pydantic import BaseModel
|
||||
|
||||
from .core import Invoker, Prompty, SimpleModel
|
||||
|
||||
|
||||
def _convert_messages_to_dicts(data: Any) -> Any:
|
||||
"""Recursively convert BaseMessage objects to dicts for mustache compatibility.
|
||||
|
||||
The mustache implementation only supports dict, list, and tuple traversal
|
||||
for security reasons. This converts any BaseMessage objects to dicts.
|
||||
"""
|
||||
if isinstance(data, BaseMessage):
|
||||
return data.model_dump()
|
||||
elif isinstance(data, dict):
|
||||
return {k: _convert_messages_to_dicts(v) for k, v in data.items()}
|
||||
elif isinstance(data, list):
|
||||
return [_convert_messages_to_dicts(item) for item in data]
|
||||
elif isinstance(data, tuple):
|
||||
return tuple(_convert_messages_to_dicts(item) for item in data)
|
||||
return data
|
||||
|
||||
|
||||
class MustacheRenderer(Invoker):
|
||||
"""Render a mustache template."""
|
||||
|
||||
def __init__(self, prompty: Prompty) -> None:
|
||||
self.prompty = prompty
|
||||
|
||||
def invoke(self, data: BaseModel) -> BaseModel:
|
||||
if not isinstance(data, SimpleModel):
|
||||
raise ValueError("Expected data to be an instance of SimpleModel")
|
||||
# Convert any BaseMessage objects to dicts for mustache compatibility
|
||||
converted_data = _convert_messages_to_dicts(data.item)
|
||||
generated = mustache.render(self.prompty.content, converted_data)
|
||||
return SimpleModel[str](item=generated)
|
||||
@@ -1,252 +0,0 @@
|
||||
import traceback
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from .core import (
|
||||
Frontmatter,
|
||||
InvokerFactory,
|
||||
ModelSettings,
|
||||
Prompty,
|
||||
PropertySettings,
|
||||
SimpleModel,
|
||||
TemplateSettings,
|
||||
param_hoisting,
|
||||
)
|
||||
|
||||
|
||||
def load(prompt_path: str, configuration: str = "default") -> Prompty:
|
||||
"""Load a prompty file and return a Prompty object.
|
||||
|
||||
Args:
|
||||
prompt_path: The path to the prompty file.
|
||||
configuration: The configuration to use. Defaults to `'default'`.
|
||||
|
||||
Returns:
|
||||
The Prompty object.
|
||||
"""
|
||||
file_path = Path(prompt_path)
|
||||
if not file_path.is_absolute():
|
||||
# get caller's path (take into account trace frame)
|
||||
caller = Path(traceback.extract_stack()[-3].filename)
|
||||
file_path = Path(caller.parent / file_path).resolve().absolute()
|
||||
|
||||
# load dictionary from prompty file
|
||||
matter = Frontmatter.read_file(file_path.__fspath__())
|
||||
attributes = matter["attributes"]
|
||||
content = matter["body"]
|
||||
|
||||
# normalize attribute dictionary resolve keys and files
|
||||
attributes = Prompty.normalize(attributes, file_path.parent)
|
||||
|
||||
# load global configuration
|
||||
if "model" not in attributes:
|
||||
attributes["model"] = {}
|
||||
|
||||
# pull model settings out of attributes
|
||||
try:
|
||||
model = ModelSettings(**attributes.pop("model"))
|
||||
except Exception as e:
|
||||
raise ValueError(f"Error in model settings: {e}")
|
||||
|
||||
# pull template settings
|
||||
try:
|
||||
if "template" in attributes:
|
||||
t = attributes.pop("template")
|
||||
if isinstance(t, dict):
|
||||
template = TemplateSettings(**t)
|
||||
# has to be a string denoting the type
|
||||
else:
|
||||
template = TemplateSettings(type=t, parser="prompty")
|
||||
else:
|
||||
template = TemplateSettings(type="mustache", parser="prompty")
|
||||
except Exception as e:
|
||||
raise ValueError(f"Error in template loader: {e}")
|
||||
|
||||
# formalize inputs and outputs
|
||||
if "inputs" in attributes:
|
||||
try:
|
||||
inputs = {
|
||||
k: PropertySettings(**v) for (k, v) in attributes.pop("inputs").items()
|
||||
}
|
||||
except Exception as e:
|
||||
raise ValueError(f"Error in inputs: {e}")
|
||||
else:
|
||||
inputs = {}
|
||||
if "outputs" in attributes:
|
||||
try:
|
||||
outputs = {
|
||||
k: PropertySettings(**v) for (k, v) in attributes.pop("outputs").items()
|
||||
}
|
||||
except Exception as e:
|
||||
raise ValueError(f"Error in outputs: {e}")
|
||||
else:
|
||||
outputs = {}
|
||||
|
||||
# recursive loading of base prompty
|
||||
if "base" in attributes:
|
||||
# load the base prompty from the same directory as the current prompty
|
||||
base = load(file_path.parent / attributes["base"])
|
||||
# hoist the base prompty's attributes to the current prompty
|
||||
model.api = base.model.api if model.api == "" else model.api
|
||||
model.configuration = param_hoisting(
|
||||
model.configuration, base.model.configuration
|
||||
)
|
||||
model.parameters = param_hoisting(model.parameters, base.model.parameters)
|
||||
model.response = param_hoisting(model.response, base.model.response)
|
||||
attributes["sample"] = param_hoisting(attributes, base.sample, "sample")
|
||||
|
||||
p = Prompty(
|
||||
**attributes,
|
||||
model=model,
|
||||
inputs=inputs,
|
||||
outputs=outputs,
|
||||
template=template,
|
||||
content=content,
|
||||
file=file_path,
|
||||
basePrompty=base,
|
||||
)
|
||||
else:
|
||||
p = Prompty(
|
||||
**attributes,
|
||||
model=model,
|
||||
inputs=inputs,
|
||||
outputs=outputs,
|
||||
template=template,
|
||||
content=content,
|
||||
file=file_path,
|
||||
)
|
||||
return p
|
||||
|
||||
|
||||
def prepare(
|
||||
prompt: Prompty,
|
||||
inputs: dict[str, Any] = {},
|
||||
) -> Any:
|
||||
"""Prepare the inputs for the prompty.
|
||||
|
||||
Args:
|
||||
prompt: The Prompty object.
|
||||
inputs: The inputs to the prompty. Defaults to `{}`.
|
||||
|
||||
Returns:
|
||||
The prepared inputs.
|
||||
"""
|
||||
invoker = InvokerFactory()
|
||||
|
||||
inputs = param_hoisting(inputs, prompt.sample)
|
||||
|
||||
if prompt.template.type == "NOOP":
|
||||
render = prompt.content
|
||||
else:
|
||||
# render
|
||||
result = invoker(
|
||||
"renderer",
|
||||
prompt.template.type,
|
||||
prompt,
|
||||
SimpleModel(item=inputs),
|
||||
)
|
||||
render = result.item
|
||||
|
||||
if prompt.template.parser == "NOOP":
|
||||
result = render
|
||||
else:
|
||||
# parse
|
||||
result = invoker(
|
||||
"parser",
|
||||
f"{prompt.template.parser}.{prompt.model.api}",
|
||||
prompt,
|
||||
SimpleModel(item=result.item),
|
||||
)
|
||||
|
||||
if isinstance(result, SimpleModel):
|
||||
return result.item
|
||||
else:
|
||||
return result
|
||||
|
||||
|
||||
def run(
|
||||
prompt: Prompty,
|
||||
content: dict | list | str,
|
||||
configuration: dict[str, Any] = {},
|
||||
parameters: dict[str, Any] = {},
|
||||
raw: bool = False,
|
||||
) -> Any:
|
||||
"""Run the prompty.
|
||||
|
||||
Args:
|
||||
prompt: The Prompty object.
|
||||
content: The content to run the prompty on.
|
||||
configuration: The configuration to use. Defaults to `{}`.
|
||||
parameters: The parameters to use. Defaults to `{}`.
|
||||
raw: Whether to return the raw output. Defaults to `False`.
|
||||
|
||||
Returns:
|
||||
The result of running the prompty.
|
||||
"""
|
||||
invoker = InvokerFactory()
|
||||
|
||||
if configuration != {}:
|
||||
prompt.model.configuration = param_hoisting(
|
||||
configuration, prompt.model.configuration
|
||||
)
|
||||
|
||||
if parameters != {}:
|
||||
prompt.model.parameters = param_hoisting(parameters, prompt.model.parameters)
|
||||
|
||||
# execute
|
||||
result = invoker(
|
||||
"executor",
|
||||
prompt.model.configuration["type"],
|
||||
prompt,
|
||||
SimpleModel(item=content),
|
||||
)
|
||||
|
||||
# skip?
|
||||
if not raw:
|
||||
# process
|
||||
result = invoker(
|
||||
"processor",
|
||||
prompt.model.configuration["type"],
|
||||
prompt,
|
||||
result,
|
||||
)
|
||||
|
||||
if isinstance(result, SimpleModel):
|
||||
return result.item
|
||||
else:
|
||||
return result
|
||||
|
||||
|
||||
def execute(
|
||||
prompt: str | Prompty,
|
||||
configuration: dict[str, Any] = {},
|
||||
parameters: dict[str, Any] = {},
|
||||
inputs: dict[str, Any] = {},
|
||||
raw: bool = False,
|
||||
connection: str = "default",
|
||||
) -> Any:
|
||||
"""Execute a `Prompty`.
|
||||
|
||||
Args:
|
||||
prompt: The prompt to execute.
|
||||
Can be a path to a prompty file or a `Prompty` object.
|
||||
configuration: The configuration to use.
|
||||
parameters: The parameters to use.
|
||||
inputs: The inputs to the `Prompty`.
|
||||
raw: Whether to return the raw output.
|
||||
connection: The connection to use.
|
||||
|
||||
Returns:
|
||||
The result of executing the `Prompty`.
|
||||
"""
|
||||
|
||||
if isinstance(prompt, str):
|
||||
prompt = load(prompt, connection)
|
||||
|
||||
# prepare content
|
||||
content = prepare(prompt, inputs)
|
||||
|
||||
# run LLM model
|
||||
result = run(prompt, content, configuration, parameters, raw)
|
||||
|
||||
return result
|
||||
@@ -1,106 +0,0 @@
|
||||
[build-system]
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[project]
|
||||
name = "langchain-prompty"
|
||||
version = "1.0.0"
|
||||
description = "An integration package connecting Prompty and LangChain"
|
||||
license = { text = "MIT" }
|
||||
readme = "README.md"
|
||||
classifiers = [
|
||||
"Development Status :: 5 - Production/Stable",
|
||||
"Intended Audience :: Developers",
|
||||
"License :: OSI Approved :: MIT License",
|
||||
"Programming Language :: Python :: 3",
|
||||
"Programming Language :: Python :: 3.10",
|
||||
"Programming Language :: Python :: 3.11",
|
||||
"Programming Language :: Python :: 3.12",
|
||||
"Programming Language :: Python :: 3.13",
|
||||
"Programming Language :: Python :: 3.14",
|
||||
"Topic :: Scientific/Engineering :: Artificial Intelligence",
|
||||
]
|
||||
requires-python = ">=3.10.0,<4.0.0"
|
||||
dependencies = [
|
||||
"langchain-core>=1.0.0,<2.0.0",
|
||||
"pyyaml>=6.0.1,<7.0.0"
|
||||
]
|
||||
|
||||
[project.urls]
|
||||
Homepage = "https://docs.langchain.com/oss/python/integrations/providers/microsoft"
|
||||
Documentation = "https://reference.langchain.com/python/integrations/langchain_prompty/"
|
||||
Repository = "https://github.com/langchain-ai/langchain"
|
||||
Issues = "https://github.com/langchain-ai/langchain/issues"
|
||||
Changelog = "https://github.com/langchain-ai/langchain/releases?q=%22langchain-prompty%22"
|
||||
Twitter = "https://x.com/LangChain"
|
||||
Slack = "https://www.langchain.com/join-community"
|
||||
Reddit = "https://www.reddit.com/r/LangChain/"
|
||||
|
||||
[dependency-groups]
|
||||
test = [
|
||||
"pytest>=7.3.0,<8.0.0",
|
||||
"pytest-mock>=3.10.0,<4.0.0",
|
||||
"pytest-watcher>=0.3.4,<1.0.0",
|
||||
"pytest-asyncio>=0.21.1,<1.0.0",
|
||||
"pytest-benchmark",
|
||||
"freezegun>=1.2.2,<2.0.0",
|
||||
"syrupy>=4.0.2,<5.0.0",
|
||||
"langchain-core",
|
||||
"langchain-classic",
|
||||
"langchain-text-splitters",
|
||||
"langchain-tests",
|
||||
]
|
||||
test_integration = []
|
||||
lint = [
|
||||
"ruff>=0.13.1,<0.14.0",
|
||||
"types-urllib3>=1.26.25.14,<2.0.0.0",
|
||||
]
|
||||
dev = [
|
||||
"types-pyyaml>=6.0.12.20240311,<7.0.0.0",
|
||||
"langchain-core"
|
||||
]
|
||||
typing = [
|
||||
"mypy>=1.18.1,<1.19.0",
|
||||
"types-pyyaml>=6.0.12.20240311,<7.0.0.0",
|
||||
"langchain-core",
|
||||
]
|
||||
|
||||
|
||||
[tool.uv.sources]
|
||||
langchain-core = { path = "../../core", editable = true }
|
||||
langchain-text-splitters = { path = "../../text-splitters", editable = true }
|
||||
langchain-classic = { path = "../../langchain", editable = true }
|
||||
langchain-tests = { path = "../../standard-tests", editable = true }
|
||||
|
||||
[tool.ruff.format]
|
||||
docstring-code-format = true
|
||||
|
||||
[tool.ruff.lint]
|
||||
select = ["E", "F", "I", "T201", "UP", "S"]
|
||||
|
||||
[tool.ruff.lint.pydocstyle]
|
||||
convention = "google"
|
||||
ignore-var-parameters = true # ignore missing documentation for *args and **kwargs parameters
|
||||
|
||||
[tool.ruff.lint.flake8-tidy-imports]
|
||||
ban-relative-imports = "all"
|
||||
|
||||
[tool.mypy]
|
||||
disallow_untyped_defs = "True"
|
||||
|
||||
[tool.coverage.run]
|
||||
omit = ["tests/*"]
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
addopts = "--snapshot-warn-unused --strict-markers --strict-config --durations=5"
|
||||
markers = [
|
||||
"requires: mark tests as requiring a specific library",
|
||||
"compile: mark placeholder test used to compile integration tests without running them",
|
||||
]
|
||||
asyncio_mode = "auto"
|
||||
|
||||
[tool.ruff.lint.extend-per-file-ignores]
|
||||
"tests/**/*.py" = [
|
||||
"S101", # Tests need assertions
|
||||
"S311", # Standard pseudo-random generators are not suitable for cryptographic purposes
|
||||
]
|
||||
@@ -1,15 +0,0 @@
|
||||
import sys
|
||||
import traceback
|
||||
from importlib.machinery import SourceFileLoader
|
||||
|
||||
if __name__ == "__main__":
|
||||
files = sys.argv[1:]
|
||||
has_failure = False
|
||||
for file in files:
|
||||
try:
|
||||
SourceFileLoader("x", file).load_module()
|
||||
except Exception:
|
||||
has_failure = True
|
||||
traceback.print_exc()
|
||||
|
||||
sys.exit(1 if has_failure else 0)
|
||||
@@ -1,17 +0,0 @@
|
||||
#!/bin/bash
|
||||
|
||||
set -eu
|
||||
|
||||
# Initialize a variable to keep track of errors
|
||||
errors=0
|
||||
|
||||
# make sure not importing from langchain or langchain_experimental
|
||||
git --no-pager grep '^from langchain\.' . && errors=$((errors+1))
|
||||
git --no-pager grep '^from langchain_experimental\.' . && errors=$((errors+1))
|
||||
|
||||
# Decide on an exit status based on the errors
|
||||
if [ "$errors" -gt 0 ]; then
|
||||
exit 1
|
||||
else
|
||||
exit 0
|
||||
fi
|
||||
Whitespace-only changes.
Whitespace-only changes.
@@ -1,7 +0,0 @@
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.mark.compile
|
||||
def test_placeholder() -> None:
|
||||
"""Used for compiling integration tests without running any real tests."""
|
||||
pass
|
||||
Whitespace-only changes.
@@ -1,397 +0,0 @@
|
||||
"""A fake callback handler for testing purposes."""
|
||||
|
||||
from itertools import chain
|
||||
from typing import Any
|
||||
from uuid import UUID
|
||||
|
||||
from langchain_core.callbacks import AsyncCallbackHandler, BaseCallbackHandler
|
||||
from langchain_core.messages import BaseMessage
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class BaseFakeCallbackHandler(BaseModel):
|
||||
"""Base fake callback handler for testing."""
|
||||
|
||||
starts: int = 0
|
||||
ends: int = 0
|
||||
errors: int = 0
|
||||
text: int = 0
|
||||
ignore_llm_: bool = False
|
||||
ignore_chain_: bool = False
|
||||
ignore_agent_: bool = False
|
||||
ignore_retriever_: bool = False
|
||||
ignore_chat_model_: bool = False
|
||||
|
||||
# to allow for similar callback handlers that are not technically equal
|
||||
fake_id: str | None = None
|
||||
|
||||
# add finer-grained counters for easier debugging of failing tests
|
||||
chain_starts: int = 0
|
||||
chain_ends: int = 0
|
||||
llm_starts: int = 0
|
||||
llm_ends: int = 0
|
||||
llm_streams: int = 0
|
||||
tool_starts: int = 0
|
||||
tool_ends: int = 0
|
||||
agent_actions: int = 0
|
||||
agent_ends: int = 0
|
||||
chat_model_starts: int = 0
|
||||
retriever_starts: int = 0
|
||||
retriever_ends: int = 0
|
||||
retriever_errors: int = 0
|
||||
retries: int = 0
|
||||
|
||||
input_prompts: list[str] = []
|
||||
|
||||
|
||||
class BaseFakeCallbackHandlerMixin(BaseFakeCallbackHandler):
|
||||
"""Base fake callback handler mixin for testing."""
|
||||
|
||||
def on_llm_start_common(self) -> None:
|
||||
self.llm_starts += 1
|
||||
self.starts += 1
|
||||
|
||||
def on_llm_end_common(self) -> None:
|
||||
self.llm_ends += 1
|
||||
self.ends += 1
|
||||
|
||||
def on_llm_error_common(self) -> None:
|
||||
self.errors += 1
|
||||
|
||||
def on_llm_new_token_common(self) -> None:
|
||||
self.llm_streams += 1
|
||||
|
||||
def on_retry_common(self) -> None:
|
||||
self.retries += 1
|
||||
|
||||
def on_chain_start_common(self) -> None:
|
||||
self.chain_starts += 1
|
||||
self.starts += 1
|
||||
|
||||
def on_chain_end_common(self) -> None:
|
||||
self.chain_ends += 1
|
||||
self.ends += 1
|
||||
|
||||
def on_chain_error_common(self) -> None:
|
||||
self.errors += 1
|
||||
|
||||
def on_tool_start_common(self) -> None:
|
||||
self.tool_starts += 1
|
||||
self.starts += 1
|
||||
|
||||
def on_tool_end_common(self) -> None:
|
||||
self.tool_ends += 1
|
||||
self.ends += 1
|
||||
|
||||
def on_tool_error_common(self) -> None:
|
||||
self.errors += 1
|
||||
|
||||
def on_agent_action_common(self) -> None:
|
||||
self.agent_actions += 1
|
||||
self.starts += 1
|
||||
|
||||
def on_agent_finish_common(self) -> None:
|
||||
self.agent_ends += 1
|
||||
self.ends += 1
|
||||
|
||||
def on_chat_model_start_common(self) -> None:
|
||||
self.chat_model_starts += 1
|
||||
self.starts += 1
|
||||
|
||||
def on_text_common(self) -> None:
|
||||
self.text += 1
|
||||
|
||||
def on_retriever_start_common(self) -> None:
|
||||
self.starts += 1
|
||||
self.retriever_starts += 1
|
||||
|
||||
def on_retriever_end_common(self) -> None:
|
||||
self.ends += 1
|
||||
self.retriever_ends += 1
|
||||
|
||||
def on_retriever_error_common(self) -> None:
|
||||
self.errors += 1
|
||||
self.retriever_errors += 1
|
||||
|
||||
|
||||
class FakeCallbackHandler(BaseCallbackHandler, BaseFakeCallbackHandlerMixin):
|
||||
"""Fake callback handler for testing."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.input_prompts = []
|
||||
|
||||
@property
|
||||
def ignore_llm(self) -> bool:
|
||||
"""Whether to ignore LLM callbacks."""
|
||||
return self.ignore_llm_
|
||||
|
||||
@property
|
||||
def ignore_chain(self) -> bool:
|
||||
"""Whether to ignore chain callbacks."""
|
||||
return self.ignore_chain_
|
||||
|
||||
@property
|
||||
def ignore_agent(self) -> bool:
|
||||
"""Whether to ignore agent callbacks."""
|
||||
return self.ignore_agent_
|
||||
|
||||
@property
|
||||
def ignore_retriever(self) -> bool:
|
||||
"""Whether to ignore retriever callbacks."""
|
||||
return self.ignore_retriever_
|
||||
|
||||
def on_llm_start(
|
||||
self, serialized: dict[str, Any], prompts: list[str], **kwargs: Any
|
||||
) -> Any:
|
||||
self.input_prompts = prompts
|
||||
self.on_llm_start_common()
|
||||
|
||||
def on_llm_new_token(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
self.on_llm_new_token_common()
|
||||
|
||||
def on_llm_end(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
self.on_llm_end_common()
|
||||
|
||||
def on_llm_error(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
self.on_llm_error_common()
|
||||
|
||||
def on_retry(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
self.on_retry_common()
|
||||
|
||||
def on_chain_start(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
self.on_chain_start_common()
|
||||
|
||||
def on_chain_end(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
self.on_chain_end_common()
|
||||
|
||||
def on_chain_error(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
self.on_chain_error_common()
|
||||
|
||||
def on_tool_start(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
self.on_tool_start_common()
|
||||
|
||||
def on_tool_end(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
self.on_tool_end_common()
|
||||
|
||||
def on_tool_error(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
self.on_tool_error_common()
|
||||
|
||||
def on_agent_action(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
self.on_agent_action_common()
|
||||
|
||||
def on_agent_finish(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
self.on_agent_finish_common()
|
||||
|
||||
def on_text(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
self.on_text_common()
|
||||
|
||||
def on_retriever_start(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
self.on_retriever_start_common()
|
||||
|
||||
def on_retriever_end(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
self.on_retriever_end_common()
|
||||
|
||||
def on_retriever_error(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
self.on_retriever_error_common()
|
||||
|
||||
# Overriding since BaseModel has __deepcopy__ method as well
|
||||
def __deepcopy__(self, memo: dict) -> "FakeCallbackHandler": # type: ignore
|
||||
return self
|
||||
|
||||
|
||||
class FakeCallbackHandlerWithChatStart(FakeCallbackHandler):
|
||||
def on_chat_model_start(
|
||||
self,
|
||||
serialized: dict[str, Any],
|
||||
messages: list[list[BaseMessage]],
|
||||
*,
|
||||
run_id: UUID,
|
||||
parent_run_id: UUID | None = None,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
assert all(isinstance(m, BaseMessage) for m in chain(*messages))
|
||||
self.on_chat_model_start_common()
|
||||
|
||||
|
||||
class FakeAsyncCallbackHandler(AsyncCallbackHandler, BaseFakeCallbackHandlerMixin):
|
||||
"""Fake async callback handler for testing."""
|
||||
|
||||
@property
|
||||
def ignore_llm(self) -> bool:
|
||||
"""Whether to ignore LLM callbacks."""
|
||||
return self.ignore_llm_
|
||||
|
||||
@property
|
||||
def ignore_chain(self) -> bool:
|
||||
"""Whether to ignore chain callbacks."""
|
||||
return self.ignore_chain_
|
||||
|
||||
@property
|
||||
def ignore_agent(self) -> bool:
|
||||
"""Whether to ignore agent callbacks."""
|
||||
return self.ignore_agent_
|
||||
|
||||
async def on_retry(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
self.on_retry_common()
|
||||
|
||||
async def on_llm_start(
|
||||
self, serialized: dict[str, Any], prompts: list[str], **kwargs: Any
|
||||
) -> None:
|
||||
self.on_llm_start_common()
|
||||
|
||||
async def on_llm_new_token(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
self.on_llm_new_token_common()
|
||||
|
||||
async def on_llm_end(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
self.on_llm_end_common()
|
||||
|
||||
async def on_llm_error(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
self.on_llm_error_common()
|
||||
|
||||
async def on_chain_start(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
self.on_chain_start_common()
|
||||
|
||||
async def on_chain_end(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
self.on_chain_end_common()
|
||||
|
||||
async def on_chain_error(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
self.on_chain_error_common()
|
||||
|
||||
async def on_tool_start(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
self.on_tool_start_common()
|
||||
|
||||
async def on_tool_end(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
self.on_tool_end_common()
|
||||
|
||||
async def on_tool_error(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
self.on_tool_error_common()
|
||||
|
||||
async def on_agent_action(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
self.on_agent_action_common()
|
||||
|
||||
async def on_agent_finish(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
self.on_agent_finish_common()
|
||||
|
||||
async def on_text(
|
||||
self,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
self.on_text_common()
|
||||
|
||||
# Overriding since BaseModel has __deepcopy__ method as well
|
||||
def __deepcopy__(self, memo: dict) -> "FakeAsyncCallbackHandler": # type: ignore
|
||||
return self
|
||||
@@ -1,45 +0,0 @@
|
||||
"""Fake Chat Model wrapper for testing purposes."""
|
||||
|
||||
import json
|
||||
from typing import Any
|
||||
|
||||
from langchain_core.callbacks import (
|
||||
AsyncCallbackManagerForLLMRun,
|
||||
CallbackManagerForLLMRun,
|
||||
)
|
||||
from langchain_core.language_models.chat_models import SimpleChatModel
|
||||
from langchain_core.messages import AIMessage, BaseMessage
|
||||
from langchain_core.outputs import ChatGeneration, ChatResult
|
||||
|
||||
|
||||
class FakeEchoPromptChatModel(SimpleChatModel):
|
||||
"""Fake Chat Model wrapper for testing purposes."""
|
||||
|
||||
def _call(
|
||||
self,
|
||||
messages: list[BaseMessage],
|
||||
stop: list[str] | None = None,
|
||||
run_manager: CallbackManagerForLLMRun | None = None,
|
||||
**kwargs: Any,
|
||||
) -> str:
|
||||
return json.dumps([message.model_dump() for message in messages])
|
||||
|
||||
async def _agenerate(
|
||||
self,
|
||||
messages: list[BaseMessage],
|
||||
stop: list[str] | None = None,
|
||||
run_manager: AsyncCallbackManagerForLLMRun | None = None,
|
||||
**kwargs: Any,
|
||||
) -> ChatResult:
|
||||
output_str = "fake response 2"
|
||||
message = AIMessage(content=output_str)
|
||||
generation = ChatGeneration(message=message)
|
||||
return ChatResult(generations=[generation])
|
||||
|
||||
@property
|
||||
def _llm_type(self) -> str:
|
||||
return "fake-echo-prompt-chat-model"
|
||||
|
||||
@property
|
||||
def _identifying_params(self) -> dict[str, Any]:
|
||||
return {"key": "fake"}
|
||||
@@ -1,40 +0,0 @@
|
||||
from langchain_classic.agents import AgentOutputParser
|
||||
from langchain_core.agents import AgentAction, AgentFinish
|
||||
|
||||
|
||||
def extract_action_details(text: str) -> tuple[str | None, str | None]:
|
||||
# Split the text into lines and strip whitespace
|
||||
lines = [line.strip() for line in text.strip().split("\n")]
|
||||
|
||||
# Initialize variables to hold the extracted values
|
||||
action = None
|
||||
action_input = None
|
||||
|
||||
# Iterate through the lines to find and extract the desired information
|
||||
for line in lines:
|
||||
if line.startswith("Action:"):
|
||||
action = line.split(":", 1)[1].strip()
|
||||
elif line.startswith("Action Input:"):
|
||||
action_input = line.split(":", 1)[1].strip()
|
||||
|
||||
return action, action_input
|
||||
|
||||
|
||||
class FakeOutputParser(AgentOutputParser):
|
||||
def parse(self, text: str) -> AgentAction | AgentFinish:
|
||||
action, input = extract_action_details(text)
|
||||
|
||||
if action:
|
||||
log = f"\nInvoking: `{action}` with `{input}"
|
||||
|
||||
return AgentAction(tool=action, tool_input=(input or ""), log=log)
|
||||
elif "Final Answer" in text:
|
||||
return AgentFinish({"output": text}, text)
|
||||
|
||||
return AgentAction(
|
||||
"Intermediate Answer", "after_colon", "Final Answer: This should end"
|
||||
)
|
||||
|
||||
@property
|
||||
def _type(self) -> str:
|
||||
return "self_ask"
|
||||
Whitespace-only changes.
@@ -1,28 +0,0 @@
|
||||
---
|
||||
name: Basic Prompt
|
||||
description: A basic prompt that uses the GPT-3 chat API to answer questions
|
||||
authors:
|
||||
- author_1
|
||||
- author_2
|
||||
model:
|
||||
api: chat
|
||||
configuration:
|
||||
azure_deployment: gpt-35-turbo
|
||||
sample:
|
||||
firstName: Jane
|
||||
lastName: Doe
|
||||
question: What is the meaning of life?
|
||||
chat_history: []
|
||||
---
|
||||
system:
|
||||
You are an AI assistant who helps people find information.
|
||||
As the assistant, you answer questions briefly, succinctly,
|
||||
and in a personable manner using markdown and even add some personal flair with appropriate emojis.
|
||||
|
||||
{{#chat_history}}
|
||||
{{role}}:
|
||||
{{content}}
|
||||
{{/chat_history}}
|
||||
|
||||
user:
|
||||
{{input}}
|
||||
@@ -1,32 +0,0 @@
|
||||
---
|
||||
name: Basic Prompt
|
||||
description: A basic prompt that uses the GPT-3 chat API to answer questions
|
||||
authors:
|
||||
- author_1
|
||||
- author_2
|
||||
model:
|
||||
api: chat
|
||||
configuration:
|
||||
azure_deployment: gpt-35-turbo
|
||||
sample:
|
||||
firstName: Jane
|
||||
lastName: Doe
|
||||
input: What is the meaning of life?
|
||||
chat_history: []
|
||||
---
|
||||
system:
|
||||
You are an AI assistant who helps people find information.
|
||||
As the assistant, you answer questions briefly, succinctly,
|
||||
and in a personable manner using markdown and even add some personal flair with appropriate emojis.
|
||||
|
||||
# Customer
|
||||
You are helping {{firstName}} {{lastName}} to find answers to their questions.
|
||||
Use their name to address them in your responses.
|
||||
|
||||
{{#chat_history}}
|
||||
{{type}}:
|
||||
{{content}}
|
||||
{{/chat_history}}
|
||||
|
||||
user:
|
||||
{{input}}
|
||||
@@ -1,10 +0,0 @@
|
||||
---
|
||||
name: IssuePrompt
|
||||
description: A prompt used to detect if double templating occurs
|
||||
model:
|
||||
api: chat
|
||||
template: mustache
|
||||
---
|
||||
|
||||
user:
|
||||
{{user_input}}
|
||||
@@ -1,7 +0,0 @@
|
||||
from langchain_prompty import __all__
|
||||
|
||||
EXPECTED_ALL = ["create_chat_prompt"]
|
||||
|
||||
|
||||
def test_all_imports() -> None:
|
||||
assert sorted(EXPECTED_ALL) == sorted(__all__)
|
||||
@@ -1,62 +0,0 @@
|
||||
"""Test thread-safety of InvokerFactory singleton pattern."""
|
||||
|
||||
import threading
|
||||
|
||||
from langchain_prompty.core import InvokerFactory
|
||||
|
||||
|
||||
def test_invoker_factory_singleton_thread_safety() -> None:
|
||||
"""Test that InvokerFactory maintains singleton pattern under concurrent access.
|
||||
|
||||
This test verifies the fix for issue #34981 by ensuring that multiple threads
|
||||
accessing InvokerFactory simultaneously all receive the same instance.
|
||||
"""
|
||||
# Reset singleton for testing
|
||||
InvokerFactory._instance = None
|
||||
|
||||
results: list[int] = []
|
||||
|
||||
def thread_worker() -> None:
|
||||
"""Worker function that gets InvokerFactory instance."""
|
||||
factory = InvokerFactory()
|
||||
results.append(id(factory))
|
||||
|
||||
# Create 10 threads to access the singleton concurrently
|
||||
threads = [threading.Thread(target=thread_worker) for _ in range(10)]
|
||||
|
||||
# Start all threads
|
||||
for thread in threads:
|
||||
thread.start()
|
||||
|
||||
# Wait for all threads to complete
|
||||
for thread in threads:
|
||||
thread.join()
|
||||
|
||||
# Verify all threads received the same instance
|
||||
unique_instances = set(results)
|
||||
assert len(unique_instances) == 1, (
|
||||
f"Expected 1 unique instance, but got {len(unique_instances)}. "
|
||||
f"Race condition detected!"
|
||||
)
|
||||
|
||||
|
||||
def test_invoker_factory_singleton_consistency() -> None:
|
||||
"""Test that multiple sequential calls return the same instance."""
|
||||
factory1 = InvokerFactory()
|
||||
factory2 = InvokerFactory()
|
||||
factory3 = InvokerFactory()
|
||||
|
||||
assert factory1 is factory2
|
||||
assert factory2 is factory3
|
||||
assert id(factory1) == id(factory2) == id(factory3)
|
||||
|
||||
|
||||
def test_invoker_factory_noop_registrations() -> None:
|
||||
"""Test that NOOP invokers are properly registered."""
|
||||
factory = InvokerFactory()
|
||||
|
||||
# Verify NOOP invokers are registered
|
||||
assert "NOOP" in factory._renderers
|
||||
assert "NOOP" in factory._parsers
|
||||
assert "NOOP" in factory._executors
|
||||
assert "NOOP" in factory._processors
|
||||
@@ -1,163 +0,0 @@
|
||||
import json
|
||||
import os
|
||||
|
||||
from langchain_classic.agents.format_scratchpad import (
|
||||
format_to_openai_function_messages,
|
||||
)
|
||||
from langchain_classic.tools import tool
|
||||
from langchain_core.language_models import FakeListLLM
|
||||
from langchain_core.messages import AIMessage, HumanMessage
|
||||
from langchain_core.utils.function_calling import convert_to_openai_function
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
import langchain_prompty
|
||||
|
||||
from .fake_callback_handler import FakeCallbackHandler
|
||||
from .fake_chat_model import FakeEchoPromptChatModel
|
||||
from .fake_output_parser import FakeOutputParser
|
||||
|
||||
prompty_folder_relative = "./prompts/"
|
||||
# Get the directory of the current script
|
||||
current_script_dir = os.path.dirname(__file__)
|
||||
|
||||
# Combine the current script directory with the relative path
|
||||
prompty_folder = os.path.abspath(
|
||||
os.path.join(current_script_dir, prompty_folder_relative)
|
||||
)
|
||||
|
||||
|
||||
def test_prompty_basic_chain() -> None:
|
||||
prompt = langchain_prompty.create_chat_prompt(f"{prompty_folder}/chat.prompty")
|
||||
model = FakeEchoPromptChatModel()
|
||||
chain = prompt | model
|
||||
|
||||
parsed_prompts = chain.invoke(
|
||||
{
|
||||
"firstName": "fakeFirstName",
|
||||
"lastName": "fakeLastName",
|
||||
"input": "fakeQuestion",
|
||||
}
|
||||
)
|
||||
|
||||
if isinstance(parsed_prompts.content, str):
|
||||
msgs = json.loads(str(parsed_prompts.content))
|
||||
else:
|
||||
msgs = parsed_prompts.content
|
||||
|
||||
assert len(msgs) == 2
|
||||
# Test for system and user entries
|
||||
system_message = msgs[0]
|
||||
user_message = msgs[1]
|
||||
|
||||
# Check the types of the messages
|
||||
assert system_message["type"] == "system", (
|
||||
"The first message should be of type 'system'."
|
||||
)
|
||||
assert user_message["type"] == "human", (
|
||||
"The second message should be of type 'human'."
|
||||
)
|
||||
|
||||
# Test for existence of fakeFirstName and fakeLastName in the system message
|
||||
assert "fakeFirstName" in system_message["content"], (
|
||||
"The string 'fakeFirstName' should be in the system message content."
|
||||
)
|
||||
assert "fakeLastName" in system_message["content"], (
|
||||
"The string 'fakeLastName' should be in the system message content."
|
||||
)
|
||||
|
||||
# Test for existence of fakeQuestion in the user message
|
||||
assert "fakeQuestion" in user_message["content"], (
|
||||
"The string 'fakeQuestion' should be in the user message content."
|
||||
)
|
||||
|
||||
|
||||
def test_prompty_used_in_agent() -> None:
|
||||
prompt = langchain_prompty.create_chat_prompt(f"{prompty_folder}/chat.prompty")
|
||||
tool_name = "search"
|
||||
responses = [
|
||||
f"FooBarBaz\nAction: {tool_name}\nAction Input: fakeSearch",
|
||||
"Oh well\nFinal Answer: fakefinalresponse",
|
||||
]
|
||||
callbackHandler = FakeCallbackHandler()
|
||||
llm = FakeListLLM(responses=responses, callbacks=[callbackHandler])
|
||||
|
||||
@tool
|
||||
def search(query: str) -> str:
|
||||
"""Look up things."""
|
||||
return "FakeSearchResponse"
|
||||
|
||||
tools = [search]
|
||||
llm_with_tools = llm.bind(functions=[convert_to_openai_function(t) for t in tools])
|
||||
|
||||
agent = (
|
||||
{ # type: ignore[var-annotated]
|
||||
"firstName": lambda x: x["firstName"],
|
||||
"lastName": lambda x: x["lastName"],
|
||||
"input": lambda x: x["input"],
|
||||
"chat_history": lambda x: x["chat_history"],
|
||||
"agent_scratchpad": lambda x: (
|
||||
format_to_openai_function_messages(x["intermediate_steps"])
|
||||
if "intermediate_steps" in x
|
||||
else []
|
||||
),
|
||||
}
|
||||
| prompt
|
||||
| llm_with_tools
|
||||
| FakeOutputParser()
|
||||
)
|
||||
|
||||
from langchain_classic.agents import AgentExecutor
|
||||
|
||||
class AgentInput(BaseModel):
|
||||
input: str
|
||||
chat_history: list[tuple[str, str]] = Field(
|
||||
...,
|
||||
json_schema_extra={
|
||||
"widget": {"type": "chat", "input": "input", "output": "output"}
|
||||
},
|
||||
)
|
||||
|
||||
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True).with_types(
|
||||
input_type=AgentInput # type: ignore[arg-type]
|
||||
)
|
||||
|
||||
agent_executor.invoke(
|
||||
{
|
||||
"firstName": "fakeFirstName",
|
||||
"lastName": "fakeLastName",
|
||||
"input": "fakeQuestion",
|
||||
"chat_history": [
|
||||
AIMessage(content="chat_history_1_ai"),
|
||||
HumanMessage(content="chat_history_1_human"),
|
||||
],
|
||||
}
|
||||
)
|
||||
input_prompt = callbackHandler.input_prompts[0]
|
||||
|
||||
# Test for existence of fakeFirstName and fakeLastName in the system message
|
||||
assert "fakeFirstName" in input_prompt
|
||||
assert "fakeLastName" in input_prompt
|
||||
assert "chat_history_1_ai" in input_prompt
|
||||
assert "chat_history_1_human" in input_prompt
|
||||
assert "fakeQuestion" in input_prompt
|
||||
assert "fakeSearch" in input_prompt
|
||||
|
||||
|
||||
def test_all_prompty_can_run() -> None:
|
||||
exclusions = ["embedding.prompty", "groundedness.prompty"]
|
||||
|
||||
prompty_files = [
|
||||
f
|
||||
for f in os.listdir(prompty_folder)
|
||||
if os.path.isfile(os.path.join(prompty_folder, f))
|
||||
and f.endswith(".prompty")
|
||||
and f not in exclusions
|
||||
]
|
||||
|
||||
for file in prompty_files:
|
||||
file_path = os.path.join(prompty_folder, file)
|
||||
|
||||
prompt = langchain_prompty.create_chat_prompt(file_path)
|
||||
model = FakeEchoPromptChatModel()
|
||||
chain = prompt | model
|
||||
chain.invoke({})
|
||||
@@ -1,15 +0,0 @@
|
||||
import pytest
|
||||
from pytest_benchmark.fixture import BenchmarkFixture # type: ignore[import]
|
||||
|
||||
from langchain_prompty import create_chat_prompt
|
||||
|
||||
|
||||
@pytest.mark.benchmark
|
||||
def test_create_chat_prompt_init_time(benchmark: BenchmarkFixture) -> None:
|
||||
"""Test create_chat_prompt initialization time."""
|
||||
|
||||
def _create_chat_prompts() -> None:
|
||||
for _ in range(10):
|
||||
create_chat_prompt("Hello world")
|
||||
|
||||
benchmark(_create_chat_prompts)
|
||||
@@ -1,23 +0,0 @@
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from langchain_prompty import create_chat_prompt
|
||||
|
||||
PROMPT_DIR = Path(__file__).parent / "prompts"
|
||||
|
||||
|
||||
def test_double_templating() -> None:
|
||||
"""
|
||||
Assess whether double templating occurs when invoking a chat prompt.
|
||||
If it does, an error is thrown and the test fails.
|
||||
"""
|
||||
|
||||
prompt_path = PROMPT_DIR / "double_templating.prompty"
|
||||
templated_prompt = create_chat_prompt(str(prompt_path))
|
||||
query = "What do you think of this JSON object: {'key': 7}?"
|
||||
|
||||
try:
|
||||
templated_prompt.invoke(input={"user_input": query})
|
||||
except KeyError as e:
|
||||
pytest.fail("Double templating occurred: " + str(e))
|
||||
Generated
-1866
File diff suppressed because it is too large.
Load diff
Reference in new issue
Block a user