synth_machine
v0.6.1
人工智能代理是國家機器而不是dags
合成機器可通過提供SynthDefinition來定義結構化的AI工作流來創建和運行AI Agent State Machines（ Synth ）。
國家機器是一種強大的結構，因為它們使域專家能夠將問題解構為一組狀態和過渡。
然後，狀態之間的過渡可以調用LLM，工具，數據過程或許多輸出的混合物。
安裝軟件包。 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以進行本地用途。
有用的鏈接：
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文檔或Synth_machine/Synth_definition.py中找到SynthDefinition 。構成SynthDefinition的Pydantic鹼基模型將是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）和trigger_streaming集成到實時流LLM世代。
我們提供多個執行者來生成本地或API驅動的LLM聊天完成。
openai ：https：//openai.com/api/pricing/togetherai ：https：//docs.together.ai/docs/inference-modelsanthropic ：https：//docs.anthropic.com/en/docs/models-overviewgoogle ：https：//cloud.google.com/vertex-ai/generative-ai/docs/multimodal/overview VLLM ：https：//github.com/vllm-project/vllmLlama-CPP ：https：//github.com/ggerganov/llama.cpp Model Config您可以在過渡輸出中指定default-model-config和synth base或model_config中的提供商和模型。
ModelConfig:
...
executor: [openai|togetherai|anthropic|vllm|llamacpp]
llm_name: [model_name]
代理存儲器是一個字典，包含所有臨時變量在以前的狀態和人 /系統輸入中創建的。
agent.memory
# -> {
# "[memory_key]": [memory_value]
# }
後處理功能僅應用於基本膠水代碼，所有主要功能都應置於工具中。
轉到"./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是通過為LLM試圖生成的材料提供語義上相似的示例或發揮作用，是改善LLM響應的強大工具。
synth_machine的靈活性，只要您從synth_machine.RAG繼承並創建：
embed(documents: List[str])和query(prompt: str, rag_config: Optional[synth_machine.RAGConfig])集成多個提供商和向量數據庫很容易。隨著時間的流逝，將得到支持，並在各種嵌入式提供商和向量數據庫中實現社區RAG實現。
以下抹布類是在CPU上嘗試局部抹佈設置的理想選擇。
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用於文件存儲，將ObjectStore(":memory:")作為默認值。
要檢索文件： agent.store.get(file_name)
對象存儲允許輕鬆集成到：
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
}
)
...
- key: trigger_udf
inputs:
- key: variable_key
outputs:
- key: example_udf
udf: abc
注意：任何非瑣碎功能都應是工具，而不是UDF。
