synth_machine

シンセマシン

AIエージェントは、DAGSではなく状態マシンです

シンセマシンを使用すると、ユーザーは、構造化されたAIワークフローを定義するSynthDefinitionを提供することにより、AIエージェント状態マシン（ Synth ）を作成および実行できます。
州のマシンは、ドメインの専門家が問題を状態と移行のセットに分解できるようにするため、強力な構成要素です。
状態間の遷移は、LLM、ツール、データプロセス、または多くの出力の混合を呼び出すことができます。

インストール

APIモデル

パッケージをインストールします。 pip install synth_machine[openai,togetherai,anthropic]またはpoetry add synth_machine[openai,togetherai,anthropic]

APIプロバイダー環境キーをセットアップする

 # You only need to set the API providers you want to use.
export OPENAI_API_KUY=secret
export ANTHROPIC_API_KEY=secret
export TOGETHER_API_KEY=secret

（まもなく）ローカルモデル

pip install synth_machine[vllm,llamacpp]またはpoetry add synth_machine[vllm,llamacpp]

ローカルで使用するには、CUDA、VLLM、またはllama.cppをセットアップする必要がある可能性があります。

役立つリンク：

• https://docs.vllm.ai/en/latest/getting_started/installation.html
• https://developer.nvidia.com/cuda-toolkit
• https://github.com/ggerganov/llama.cpp

シンセを定義します

 agent = Synth(
    config: dict[SynthDefinition], # Synth state machine defining states, transitions and prompts.
    tools=[], # List of tools the agent will use
    memory={}, # Any existing memory to add on top of any model_config.initial_memory
    rag_runner: Optional[RAG] = None # Define a RAG integration for your agent.
    postprocess_functions = [] # Any glue code functions
    store : ObjectStore = ObjectStore(":memory:") # Any files created by tools will automatically go to you object store    

SynthDefinition 、Synthdefinition docsまたはsynth_machine/synth_definition.pyにあります。 SynthDefinitionを構成するPydantic Basemodelsは、 Synthの最も正確な表現になります。
仕様には、主要バージョン間の更新があると予想されます。

エージェント状態と可能なトリガー

いつでも、現在の状態と次のトリガーを確認できます

 # Check state
agent.current_state()

# Triggers
agent.interfaces_for_available_triggers()

シンセを実行します

バッチ

 await agent.trigger(
    "[trigger_name]",
    params={
        "input_1": "hello"
    }
)

バッチトランジションコールは、その遷移で生成された出力変数を出力します。

ストリーミング

 await agent.streaming_trigger(
    "[trigger_name]",
    params={
        "input_1": "hello"
    }
)

ストリーミング応答は、次のイベントのいずれかを生成します。

 class YieldTasks(StrEnum):
    CHUNK = "CHUNK"
    MODEL_CONFIG = "MODEL_CONFIG"
    SET_MEMORY = "SET_MEMORY"
    SET_ACTIVE_OUTPUT = "SET_ACTIVE_OUTPUT"

• CHUNK ：LLM世代は、一度に1つのトークンでチャンクによって送信されます。
• MODEL_CONFIG ：プロバイダー固有のフロントエンドインターフェイスに現在使用されているエグゼキュータが現在使用されています。
• SET_MEMORP ：新しいメモリ変数を設定するイベントを送信します
• SET_ACTIVE_OUTPUT ：現在の遷移出力トリガーを生成します。

これにより、ユーザーはtriggerを使用して実験し、サーバーサイドイベント（SSE）とtrigger_streamingを使用してユーザーにLLM世代をリアルタイムストリームに統合できます。

LLMS

ローカルまたはAPI駆動型のLLMチャットの完了を生成するために、複数のエグゼクターを提供します。

APIモデル

• openai ：https：//openai.com/api/pricing/
• togetherai ：https：//docs.together.ai/docs/inference-models
• anthropic ：https：//docs.anthropic.com/en/docs/models-overview
• （まもなく） google ：https：//cloud.google.com/vertex-ai/generative-ai/docs/multimodal/overview

ローカル（まもなく）

• VLLM ：https：//github.com/vllm-project/vllm
• Llama-CPP ：https：//github.com/ggerganov/llama.cpp

Model Config

default-model-configのいずれかでプロバイダーとモデルを指定し、遷移出力でシンセベースまたはmodel_config指定できます。

 ModelConfig:
...
executor: [openai|togetherai|anthropic|vllm|llamacpp]
llm_name: [model_name]

メモリ

エージェントメモリは、すべての暫定変数が以前の状態および人間 /システム入力で作成する辞書です。

 agent.memory
# -> {
#   "[memory_key]": [memory_value]
# }

ツール

ポストプロセス関数は基本的な接着剤コードにのみ使用する必要があります。すべての主要な機能はツールに組み込む必要があります。

ツールはRESTAPISであり、JSON APIスキーマを提供することで追加できます

"./tools/tofuTool/api.pyに移動して、機能を表示します。

APIを開始します

 cd tools/tofuTool
poetry install
poetry run uvicorn api:app --port=5001 --reload

API仕様を取得します

 curl -X GET http://localhost:5001/openapi.json > openapi_schema.json

ツールを定義します

名前、APIエンドポイント、ツールOpenAPIスキーマのみを使用して、ツールを定義できます。

 tofu_tool = Tool(
    name="tofu_tool",
    api_endpoint="http://localhost:5001",
    api_spec=tool_spec
)

シンセマシンのぼろきれ

検索されたアウペメント生成は、LLMが生成しようとしている材料に意味的に類似した例を提供するか、LLM応答を改善するための強力なツールです。

synth_machine 、 synth_machine.RAGから継承して作成する限り、柔軟になります。

• embed(documents: List[str]) and
• query(prompt: str, rag_config: Optional[synth_machine.RAGConfig])

複数のプロバイダーとベクトルデータベースを簡単に統合できます。時間が経つにつれて、さまざまな埋め込みプロバイダーとベクトルデータベースにわたってサポートされ、コミュニティのぼろきれの実装が行われます。

ラグ例qdrant＆fastembed

次のRAGクラスは、CPUのローカルRAGセットアップを実験するのに最適です。

 pip install qdrant-client, fastembed

ラグクラスを定義します

 from synth_machine.rag import RAG
from qdrant_client import AsyncQdrantClient
from fastembed import TextEmbedding
from typing import List, Optional
from qdrant_client.models import Distance, VectorParams, PointStruct


class Qdrant(RAG):

    """
    VectorDB: Qdrant - https://github.com/qdrant/qdrant
    Embeddings: FastEmbed - https://github.com/qdrant/fastembed

    This provides fast and lightweight on-device CPU embeddings creation and 
    similarity search using Qdrant in memory.
    """
    
    def __init__(
        self,
        collection_name: str,
        embedding_model: str="BAAI/bge-small-en-v1.5",
        embedding_dimensions: int=384,
        embedding_threads: int=-1,
        qdrant_location: str=":memory:",
    ):
        self.embedding_model = TextEmbedding(
            model_name=embedding_model,
            threads=embedding_threads
        )
        self.embedding_dimensions = embedding_dimensions
        self.qdrant = AsyncQdrantClient(qdrant_location)
        self.collection_name = collection_name
    
    async def create_collection(self) -> bool:
        if await self.qdrant.collection_exists(self.collection_name):
            return True
        else:
            return await self.qdrant.create_collection(
                collection_name=self.collection_name,
                vectors_config=VectorParams(
                    size=self.embedding_dimensions, # maps to 'BAAI/bge-small-en-v1.5' model dimensions
                    distance=Distance.COSINE
                ) 
            )
    
    async def embed(self, documents: List[str], metadata: Optional[List[dict]]=None):
        if metadata and len(documents) != len(metadata):
            raise ValueError("documents and metadata must be the same length")
        embedding_list = list(
            self.embedding_model.embed(documents)
        )
        upsert_response = await self.qdrant.upsert(
            collection_name=self.collection_name,
            points=[
                PointStruct(
                    id=i,
                    vector=list(vector),
                    payload=metadata[i]
                )
                for i, vector in enumerate(embedding_list)
            ]
        )
        return upsert_response.status
        

    async def query(self, prompt: str, rag_config: RAGConfig) -> List[dict]:
        embedding = next(self.embedding_model.embed([prompt]))
            
        similar_responses = await self.qdrant.search(
            collection_name=self.collection_name,
            query_vector=embedding,
            limit=rag_config.n
        )
        return [
            point.payload for point in similar_responses
        ]

ここで、QDRANTクラスを開始し、 Synthを定義するときに提供します。

 qdrant = Qdrant(collection_name="tofu_examples")
await qdrant.create_collection()

agent = Synth(
    ...
    rag_runner=Qdrant
)

店

ツールは、さまざまなオブジェクトを返すことができます。ツールによって作成されたファイルは、自動的にagent.storeに移動します。ファイルストレージにはオブジェクトストアを使用し、 ObjectStore(":memory:")をデフォルトとして使用します。

ファイルを取得するには： agent.store.get(file_name)

簡単に統合できるオブジェクトストア：

• ローカルファイルストア
• S3
• GCS
• azure

GCSオブジェクトストアの例

 from synth_machine.machine import ObjectStore

agent = Agent(
    ...
    store=ObjectStore("gs://[bucket_name]/[prefix]))
)

ユーザー定義関数

カスタム機能は、ユーザー定義関数（UDF）として定義できます。
これらはSynth.memoryを入力として受け取り、 synth-machineの一部としてカスタム機能を実行できます。

 # Define postprocess function

from synth_machine.user_defined_functions import udf

@udf
def abc_postprocesss(memory):
    ...
    return memory["variable_key"]

agent = Synth(
  ...
  user_defined_functions = {
    "abc": abc_postprocess
  }
)

例UDF遷移構成

 ...
- key: trigger_udf
  inputs:
    - key: variable_key
  outputs:
    - key: example_udf
      udf: abc

注：非些細な機能は、UDFではなくツールである必要があります。