synth_machine

신디사이저

AI 에이전트는 DAG가 아닌 상태 기계입니다

신디사이저 머신을 통해 사용자는 구조화 된 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 세대는 한 번에 하나의 토큰으로 청크에 의해 보내집니다.
• MODEL_CONFIG : 현재 임원이 현재 모든 공급자 특정 프론트 엔드 인터페이스에 사용되고있는 수확량.
• SET_MEMORP : 새로운 메모리 변수를 설정하는 이벤트를 보냅니다
• SET_ACTIVE_OUTPUT : 현재 전환 출력 트리거를 생성합니다.

이를 통해 사용자는 trigger 사용하여 실험 한 다음 SSE (Server Side Events) 및 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
)

신디 머신 래그

검색 Augemented Generation은 Semanticly 유사한 사례를 제공하거나 LLM이 생성하려는 재료에 가해함으로써 LLM 응답을 개선하는 강력한 도구입니다.

synth_machine synth_machine.RAG 를 상속하고 만들기 위해 유연하게 유연하게 사용됩니다.

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

여러 공급자와 벡터 데이터베이스를 쉽게 통합 할 수 있습니다. 시간이 지남에 따라 다양한 임베딩 제공 업체 및 벡터 데이터베이스에서 지원 및 커뮤니티 래그 구현이있을 것입니다.

래그 예제 Qdrant & Fastembed

다음 래그 클래스는 CPU에서 로컬 래그 설정을 실험하는 데 이상적입니다.

 pip install qdrant-client, fastembed

RAG 클래스를 정의하십시오

 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를 사용하고 ObjectStore(":memory:") 기본값으로 사용합니다.

파일을 검색하려면 : agent.store.get(file_name)

EbjectStore는 쉽게 통합 할 수 있습니다.

• 로컬 파일 저장소
• S3
• GCS
• 하늘빛

예제 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가 아닌 도구 여야합니다.