WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-08-17 21:25:46 +02:00
Code Issues Packages Projects Releases Wiki Activity

First commit

Browse Source
This commit is contained in:
Nuno Campos
2023-08-09 19:50:50 +01:00
commit d0dbe3994c
13 changed files with 4362 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+173
View File
@@ -0,0 +1,173 @@
.vs/
.vscode/
.idea/
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
pip-wheel-metadata/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
docs/docs/_build/
# PyBuilder
target/
# Jupyter Notebook
.ipynb_checkpoints
notebooks/
# IPython
profile_default/
ipython_config.py
# pyenv
.python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# PEP 582; used by e.g. github.com/David-OConnor/pyflow
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.envrc
.venv
.venvs
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# macOS display setting files
.DS_Store
# Wandb directory
wandb/
# asdf tool versions
.tool-versions
/.ruff_cache/
*.pkl
*.bin
# integration test artifacts
data_map*
\[('_type', 'fake'), ('stop', None)]
# Replit files
*replit*
node_modules
docs/.yarn/
docs/node_modules/
docs/.docusaurus/
docs/.cache-loader/
docs/_dist
docs/api_reference/api_reference.rst
docs/api_reference/experimental_api_reference.rst
docs/api_reference/_build
docs/api_reference/*/
!docs/api_reference/_static/
!docs/api_reference/templates/
!docs/api_reference/themes/
docs/docs_skeleton/build
docs/docs_skeleton/node_modules
docs/docs_skeleton/yarn.lock
LICENSE
+21
View File
@@ -0,0 +1,21 @@
The MIT License
Copyright (c) Harrison Chase
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.
Makefile
+120
View File
@@ -0,0 +1,120 @@
.PHONY: all clean docs_build docs_clean docs_linkcheck api_docs_build api_docs_clean api_docs_linkcheck format lint test tests test_watch integration_tests docker_tests help extended_tests
# Default target executed when no arguments are given to make.
all: help
######################
# TESTING AND COVERAGE
######################
# Run unit tests and generate a coverage report.
coverage:
poetry run pytest --cov \
--cov-config=.coveragerc \
--cov-report xml \
--cov-report term-missing:skip-covered
######################
# DOCUMENTATION
######################
clean: docs_clean api_docs_clean
docs_build:
docs/.local_build.sh
docs_clean:
rm -r docs/_dist
docs_linkcheck:
poetry run linkchecker docs/_dist/docs_skeleton/ --ignore-url node_modules
api_docs_build:
poetry run python docs/api_reference/create_api_rst.py
cd docs/api_reference && poetry run make html
api_docs_clean:
rm -f docs/api_reference/api_reference.rst
cd docs/api_reference && poetry run make clean
api_docs_linkcheck:
poetry run linkchecker docs/api_reference/_build/html/index.html
# Define a variable for the test file path.
TEST_FILE ?= tests/unit_tests/
test:
poetry run pytest --disable-socket --allow-unix-socket $(TEST_FILE)
tests:
poetry run pytest --disable-socket --allow-unix-socket $(TEST_FILE)
extended_tests:
poetry run pytest --disable-socket --allow-unix-socket --only-extended tests/unit_tests
test_watch:
poetry run ptw --now . -- tests/unit_tests
integration_tests:
poetry run pytest tests/integration_tests
scheduled_tests:
poetry run pytest -m scheduled tests/integration_tests
docker_tests:
docker build -t my-langchain-image:test .
docker run --rm my-langchain-image:test
######################
# LINTING AND FORMATTING
######################
# Define a variable for Python and notebook files.
PYTHON_FILES=.
lint format: PYTHON_FILES=.
lint_diff format_diff: PYTHON_FILES=$(shell git diff --relative=libs/langchain --name-only --diff-filter=d master | grep -E '\.py$$|\.ipynb$$')
lint lint_diff:
poetry run mypy $(PYTHON_FILES)
poetry run black $(PYTHON_FILES) --check
poetry run ruff .
format format_diff:
poetry run black $(PYTHON_FILES)
poetry run ruff --select I --fix $(PYTHON_FILES)
spell_check:
poetry run codespell --toml pyproject.toml
spell_fix:
poetry run codespell --toml pyproject.toml -w
######################
# HELP
######################
help:
@echo '===================='
@echo '-- DOCUMENTATION --'
@echo 'clean - run docs_clean and api_docs_clean'
@echo 'docs_build - build the documentation'
@echo 'docs_clean - clean the documentation build artifacts'
@echo 'docs_linkcheck - run linkchecker on the documentation'
@echo 'api_docs_build - build the API Reference documentation'
@echo 'api_docs_clean - clean the API Reference documentation build artifacts'
@echo 'api_docs_linkcheck - run linkchecker on the API Reference documentation'
@echo '-- LINTING --'
@echo 'format - run code formatters'
@echo 'lint - run linters'
@echo 'spell_check - run codespell on the project'
@echo 'spell_fix - run codespell on the project and fix the errors'
@echo '-- TESTS --'
@echo 'coverage - run unit tests and generate coverage report'
@echo 'test - run unit tests'
@echo 'tests - run unit tests (alias for "make test")'
@echo 'test TEST_FILE=<test_file> - run all tests in file'
@echo 'extended_tests - run only extended unit tests'
@echo 'test_watch - run unit tests in watch mode'
@echo 'integration_tests - run integration tests'
@echo 'docker_tests - run unit tests in docker'
README.md
View File
examples/runnable-pubsub.py
+130
View File
@@ -0,0 +1,130 @@
from operator import itemgetter
from langchain.chat_models.openai import ChatOpenAI
from langchain.prompts import SystemMessagePromptTemplate
from langchain.schema.output_parser import StrOutputParser
from langchain.runnables.openai_functions import OpenAIFunctionsRouter
from permchain.connection_queue import InMemoryPubSubConnection
from permchain.pubsub import PubSub
from permchain.topic import Topic
drafter_prompt = (
SystemMessagePromptTemplate.from_template(
"You are an expert on turtles, who likes to write in pirate-speak. You have been tasked by your editor with drafting a 100-word article answering the following question."
)
+ "Question:\n\n{question}"
)
reviser_prompt = (
SystemMessagePromptTemplate.from_template(
"You are an expert on turtles. You have been tasked by your editor with revising the following draft, which was written by a non-expert. You may follow the editor's notes or not, as you see fit."
)
+ "Draft:\n\n{draft}"
+ "Editor's notes:\n\n{notes}"
)
editor_prompt = (
SystemMessagePromptTemplate.from_template(
"You are an editor. You have been tasked with editing the following draft, which was written by a non-expert. Please accept the draft if it is good enough to publish, or send it for revision, along with your notes to guide the revision."
)
+ "Draft:\n\n{draft}"
)
drafter_llm = ChatOpenAI(model="gpt-3.5-turbo")
editor_llm = ChatOpenAI(model="gpt-4")
# create topics
editor_inbox = Topic("editor_inbox")
reviser_inbox = Topic("reviser_inbox")
# write a first draft
drafter = (
Topic.IN.subscribe()
| {"draft": drafter_prompt | drafter_llm | StrOutputParser()}
| editor_inbox.publish()
)
# edit every draft, produce revision notes or accept
editor = (
editor_inbox.subscribe()
| editor_prompt
| editor_llm.bind(
functions=[
{
"name": "revise",
"description": "Sends the draft for revision",
"parameters": {
"type": "object",
"properties": {
"notes": {
"type": "string",
"description": "The editor's notes to guide the revision.",
},
},
},
},
{
"name": "accept",
"description": "Accepts the draft",
"parameters": {
"type": "object",
"properties": {"ready": {"const": True}},
},
},
]
)
| OpenAIFunctionsRouter(
{
"revise": (
{
"notes": itemgetter("notes"),
"draft": editor_inbox.current() | itemgetter("draft"),
"question": Topic.IN.current() | itemgetter("question"),
}
| reviser_inbox.publish()
),
"accept": editor_inbox.current() | Topic.OUT.publish(),
},
)
)
# every time revision notes are posted, revise latest draft
reviser = (
reviser_inbox.subscribe()
| {"draft": reviser_prompt | drafter_llm | StrOutputParser()}
| editor_inbox.publish()
)
web_researcher = PubSub(
processes=(drafter, editor, reviser),
connection=InMemoryPubSubConnection(),
)
for output in web_researcher.stream({"question": "What food do turtles eat?"}):
print("got output", output)
print("---done---")
# agent = PubSub(
# Channel.IN | Channel("planner"),
# Channel("executor") | executor | Channel("planner"),
# Channel("planner")
# | planner
# | {"action": Channel("executor"), "finish": Channel.OUT},
# )
# graph = (
# drafter
# | editor
# | RouterRunnable(
# {
# "send_for_revision": reviser,
# "accept_draft": lambda x: x["draft"],
# }
# )
# )
permchain/__init__.py
View File
permchain/connection.py
+35
View File
@@ -0,0 +1,35 @@
from abc import ABC, abstractmethod
import asyncio
from typing import Any, Callable
PubSubListener = Callable[[Any], None]
class PubSubConnection(ABC):
@abstractmethod
def listen(self, topic_name: str, listener: PubSubListener) -> None:
...
async def alisten(self, topic_name: str, listener: PubSubListener) -> None:
return await asyncio.get_event_loop().run_in_executor(
None, self.listen, topic_name, listener
)
@abstractmethod
def send(self, topic_name: str, message: Any) -> None:
...
async def asend(self, topic_name: str, message: Any) -> None:
return await asyncio.get_event_loop().run_in_executor(
None, self.send, topic_name, message
)
@abstractmethod
def disconnect(self, topic_name: str) -> None:
...
async def adisconnect(self, topic_name: str) -> None:
return await asyncio.get_event_loop().run_in_executor(
None, self.disconnect, topic_name
)
permchain/connection_queue.py
+24
View File
@@ -0,0 +1,24 @@
from collections import defaultdict
import threading
from typing import Any
from permchain.connection import PubSubConnection, PubSubListener
class InMemoryPubSubConnection(PubSubConnection):
def __init__(self) -> None:
self.topics = defaultdict(list)
self.lock = threading.Lock()
def listen(self, topic_name: str, listener: PubSubListener) -> None:
with self.lock:
self.topics[topic_name].append(listener)
def send(self, topic_name: str, message: Any) -> None:
with self.lock:
for listener in self.topics[topic_name]:
listener(message)
def disconnect(self, topic_name: str) -> None:
with self.lock:
self.topics[topic_name] = []
permchain/pubsub.py
+192
View File
@@ -0,0 +1,192 @@
from __future__ import annotations
import queue
from abc import ABC
from concurrent.futures import CancelledError, Future, ThreadPoolExecutor
from functools import partial
from itertools import filterfalse
from typing import (
Any,
Callable,
Iterator,
List,
Optional,
Sequence,
Set,
Tuple,
TypeVar,
)
from langchain.callbacks.manager import CallbackManager
from langchain.load.dump import dumpd
from langchain.load.serializable import Serializable
from langchain.schema.runnable import Runnable, RunnableConfig, _patch_config
from permchain.connection import PubSubConnection
from permchain.topic import INPUT_TOPIC, OUTPUT_TOPIC, RunnableSubscriber
T = TypeVar("T")
T_in = TypeVar("T_in")
T_out = TypeVar("T_out")
def partition(
pred: Callable[[T], bool], seq: Sequence[T]
) -> Tuple[Sequence[T], Sequence[T]]:
"""Partition entries into true entries and false entries.
partition(is_even, range(10)) --> 0 2 4 6 8 and 1 3 5 7 9
"""
return list(filter(pred, seq)), list(filterfalse(pred, seq))
class IterableQueue(queue.SimpleQueue):
done_sentinel = object()
def get(self, block: bool = True, timeout: float = None):
return super().get(block=block, timeout=timeout)
def __iter__(self):
return iter(self.get, self.done_sentinel)
def close(self):
self.put(self.done_sentinel)
class PubSub(Serializable, Runnable[Any, Any], ABC):
processes: Sequence[RunnableSubscriber[Any]]
connection: PubSubConnection
class Config:
arbitrary_types_allowed = True
def invoke(self, input: Any, config: Optional[RunnableConfig] = None) -> Any:
collected = []
for chunk in self.stream(input, config):
collected.append(chunk)
return collected
def stream(
self,
input: Any,
config: Optional[RunnableConfig] = None,
*,
max_concurrency: Optional[int] = None,
) -> Iterator[Any]:
input_processes, listener_processes = partition(
lambda r: r.topic.name == INPUT_TOPIC, self.processes
)
# setup callbacks
config = config or {}
callback_manager = CallbackManager.configure(
inheritable_callbacks=config.get("callbacks"),
local_callbacks=None,
verbose=False,
inheritable_tags=config.get("tags"),
local_tags=None,
inheritable_metadata=config.get("metadata"),
local_metadata=None,
)
# start the root run
run_manager = callback_manager.on_chain_start(dumpd(self), {"input": input})
with ThreadPoolExecutor(max_workers=max_concurrency) as executor:
# Track inflight futures
inflight: Set[Future] = set()
# Track exceptions
exceptions: List[Exception] = []
# Track output
output = IterableQueue()
def send(topic_name: str, message: Any) -> None:
"""Send a message to a topic. Injected into config."""
if topic_name == OUTPUT_TOPIC:
output.put(message)
else:
self.connection.send(topic_name, message)
def cleanup_run(fut: Future) -> None:
"""Cleanup after a process runs."""
inflight.remove(fut)
try:
exc = fut.exception()
except CancelledError:
exc = None
except Exception as e:
exc = e
if exc is not None:
exceptions.append(exc)
# Close output iterator if
# - all processes are done, or
# - an exception occurred
if not inflight or exc is not None:
output.close()
def run_once(process: RunnableSubscriber[Any], value: Any) -> None:
"""Run a process once."""
def get(topic_name: str) -> Any:
if topic_name == INPUT_TOPIC:
return input
elif topic_name == process.topic.name:
return value
else:
raise ValueError(
f"Cannot get value for {topic_name} in this context"
)
# Run process once in executor
fut = executor.submit(
process.invoke,
value,
config={
**_patch_config(
config, run_manager.get_child(process.topic.name)
),
"send": send,
"get": get,
},
)
# Add callback to cleanup
inflight.add(fut)
fut.add_done_callback(cleanup_run)
# Listen on all subscribed topics
for process in listener_processes:
self.connection.listen(process.topic.name, partial(run_once, process))
# Run input processes once
for process in input_processes:
run_once(process, input)
try:
# Yield output until all processes are done
final_output = None
for chunk in output:
yield chunk
if final_output is None:
final_output = chunk
else:
final_output += chunk
finally:
# Cleanup
for fut in inflight:
fut.cancel()
for process in listener_processes:
self.connection.disconnect(process.topic.name)
# Raise exceptions if any
if exceptions:
run_manager.on_chain_error(exceptions[0])
raise exceptions[0]
else:
run_manager.on_chain_end(final_output)
PubSub.update_forward_refs()
permchain/topic.py
+127
View File
@@ -0,0 +1,127 @@
from __future__ import annotations
from abc import ABC
from typing import (
Any,
Callable,
Generic,
Mapping,
Optional,
Sequence,
TypeVar,
)
from langchain.load.serializable import Serializable
from langchain.schema.runnable import (
Runnable,
RunnableBinding,
RunnableConfig,
RunnablePassthrough,
RunnableSequence,
Other,
_coerce_to_runnable,
)
from pydantic import Field
T = TypeVar("T")
T_in = TypeVar("T_in")
T_out = TypeVar("T_out")
INPUT_TOPIC = "__in__"
OUTPUT_TOPIC = "__out__"
class Topic(Serializable, Generic[T], ABC):
name: str
def __init__(self, name: str):
super().__init__(name=name)
def subscribe(self) -> RunnableSubscriber[T]:
return RunnableSubscriber(topic=self)
def current(self) -> RunnableCurrentValue[T]:
return RunnableCurrentValue(topic=self)
def publish(self) -> RunnablePublisher[T]:
return RunnablePublisher(topic=self)
def publish_each(self) -> Runnable[T, T]:
return RunnablePublisherEach(topic=self)
@classmethod
@property
def IN(cls) -> Topic[T_in]:
return cls[T_in](INPUT_TOPIC)
@classmethod
@property
def OUT(cls) -> Topic[T_out]:
return cls[T_out](OUTPUT_TOPIC)
class RunnableConfigForPubSub(RunnableConfig):
send: Callable[[str, Any], None]
get: Callable[[str], Any]
class RunnableSubscriber(RunnableBinding[T, Any]):
topic: Topic[T]
bound: Runnable[T, Any] = Field(default_factory=RunnablePassthrough)
kwargs: Mapping[str, Any] = Field(default_factory=dict)
def __or__(
self,
other: Runnable[Any, Other]
| Callable[[Any], Other]
| Mapping[str, Runnable[Any, Other] | Callable[[Any], Other]],
) -> RunnableSequence[T, Other]:
if isinstance(self.bound, RunnablePassthrough):
return RunnableSubscriber(
topic=self.topic, bound=_coerce_to_runnable(other)
)
else:
return RunnableSubscriber(topic=self.topic, bound=self.bound | other)
def __ror__(
self,
other: Runnable[Other, Any]
| Callable[[Any], Other]
| Mapping[str, Runnable[Other, Any] | Callable[[Other], Any]],
) -> RunnableSequence[Other, Any]:
raise NotImplementedError()
class RunnablePublisher(RunnablePassthrough[T]):
topic: Topic[T]
def invoke(self, input: T, config: Optional[RunnableConfigForPubSub] = None) -> T:
send = config.get("send", None)
if send is not None:
send(self.topic.name, input)
return super().invoke(input, config)
class RunnablePublisherEach(RunnablePublisher[Sequence[T]]):
topic: Topic[T]
def invoke(
self, input: Sequence[T], config: Optional[RunnableConfigForPubSub] = None
) -> Sequence[T]:
for item in input:
super().invoke(item, config)
class RunnableCurrentValue(Serializable, Runnable[Any, T]):
topic: Topic[T]
def invoke(self, input: T, config: Optional[RunnableConfigForPubSub] = None) -> T:
get = config.get("get", None)
if get is not None:
return get(self.topic.name)
else:
raise ValueError("Cannot get value in this context")
poetry.lock
Generated
+3460
View File
File diff suppressed because it is too large Load Diff
poetry.toml
+5
View File
@@ -0,0 +1,5 @@
[virtualenvs]
in-project = true
[installer]
modern-installation = false
pyproject.toml
+75
View File
@@ -0,0 +1,75 @@
[tool.poetry]
name = "permchain"
version = "0.0.1"
description = "Longchain, Langchaaaaain, etc."
authors = []
license = "MIT"
readme = "README.md"
repository = "https://www.github.com/langchain-ai/permchain"
[tool.poetry.dependencies]
python = ">=3.8.1,<4.0"
langchain = {path = "../langchain/libs/langchain", develop = true}
[tool.poetry.group.test.dependencies]
# The only dependencies that should be added are
# dependencies used for running tests (e.g., pytest, freezegun, response).
# Any dependencies that do not meet that criteria will be removed.
pytest = "^7.3.0"
pytest-cov = "^4.0.0"
pytest-dotenv = "^0.5.2"
pytest-asyncio = "^0.20.3"
pytest-mock = "^3.10.0"
syrupy = "^4.0.2"
[tool.poetry.group.lint.dependencies]
ruff = "^0.0.249"
black = "^23.1.0"
[tool.poetry.group.typing.dependencies]
mypy = "^0.991"
[tool.poetry.group.dev]
optional = true
[tool.poetry.group.dev.dependencies]
jupyter = "^1.0.0"
playwright = "^1.28.0"
setuptools = "^67.6.1"
openai = "^0.27.8"
[tool.ruff]
select = [
"E", # pycodestyle
"F", # pyflakes
"I", # isort
]
[tool.mypy]
ignore_missing_imports = "True"
disallow_untyped_defs = "True"
exclude = ["notebooks", "examples", "example_data"]
[tool.coverage.run]
omit = [
"tests/*",
]
[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"
[tool.pytest.ini_options]
# --strict-markers will raise errors on unknown marks.
# https://docs.pytest.org/en/7.1.x/how-to/mark.html#raising-errors-on-unknown-marks
#
# https://docs.pytest.org/en/7.1.x/reference/reference.html
# --strict-config any warnings encountered while parsing the `pytest`
# section of the configuration file raise errors.
#
# https://github.com/tophat/syrupy
# --snapshot-warn-unused Prints a warning on unused snapshots rather than fail the test suite.
addopts = "--strict-markers --strict-config --durations=5 --snapshot-warn-unused"
# Registering custom markers.
# https://docs.pytest.org/en/7.1.x/example/markers.html#registering-markers