synth_machine
v0.6.1
ตัวแทน AI เป็นเครื่องจักรของรัฐไม่ใช่ dags
Synth Machines ช่วยให้ผู้ใช้สร้างและเรียกใช้ AI Agent State Machines ( Synth ) โดยการจัดหา SynthDefinition เพื่อกำหนดเวิร์กโฟลว์ AI ที่มีโครงสร้าง
เครื่องจักรของรัฐเป็นโครงสร้างที่ทรงพลังเนื่องจากช่วยให้ผู้เชี่ยวชาญโดเมนสามารถแยกแยะปัญหาในชุดของรัฐและการเปลี่ยนผ่าน
การเปลี่ยนระหว่างรัฐสามารถเรียก 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 สามารถพบได้ในเอกสาร synthdefinition หรือ synth_machine/synth_definition.py basemodels pydantic ที่ประกอบขึ้นเป็น SynthDefinition จะเป็นตัวแทนที่แม่นยำที่สุดของ 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 แล้วรวมเข้ากับรุ่น LLM แบบเรียลไทม์กับผู้ใช้โดยใช้เหตุการณ์เซิร์ฟเวอร์ (SSE) และ trigger_streaming
เราเสนอผู้บริหารหลายคนเพื่อสร้างการแชท LLM ที่ขับเคลื่อนด้วย API หรือ API
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 หรือ model_config บนเอาต์พุตการเปลี่ยนแปลง
ModelConfig:
...
executor: [openai|togetherai|anthropic|vllm|llamacpp]
llm_name: [model_name]
หน่วยความจำตัวแทนเป็นพจนานุกรมที่มีตัวแปรชั่วคราวทั้งหมดสร้างขึ้นในสถานะก่อนหน้าและอินพุตของมนุษย์ / ระบบ
agent.memory
# -> {
# "[memory_key]": [memory_value]
# }
ฟังก์ชั่น Postprocess ควรใช้สำหรับรหัสกาวพื้นฐานเท่านั้นฟังก์ชั่นหลักทั้งหมดควรถูกสร้างไว้ในเครื่องมือ
ไปที่ "./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 Schema
tofu_tool = Tool(
name="tofu_tool",
api_endpoint="http://localhost:5001",
api_spec=tool_spec
)
Generation Auppremented Retrieval เป็นเครื่องมือที่มีประสิทธิภาพในการปรับปรุงการตอบสนอง LLM โดยการให้ตัวอย่างที่คล้ายคลึงกันทางความหมายหรือออกแรงกับวัสดุที่ LLM กำลังพยายามสร้าง
synth_machine มีความยืดหยุ่นเช่นเดียวกับที่คุณสืบทอดมาจาก synth_machine.RAG และสร้าง:
embed(documents: List[str]) และquery(prompt: str, rag_config: Optional[synth_machine.RAGConfig])เป็นเรื่องง่ายที่จะรวมฐานข้อมูลผู้ให้บริการและเวกเตอร์หลายแห่ง เมื่อเวลาผ่านไปจะได้รับการสนับสนุนและการใช้งาน RAG ของชุมชนในผู้ให้บริการ Embeddings และฐานข้อมูลเวกเตอร์ที่หลากหลาย
คลาส RAG ต่อไปนี้เหมาะสำหรับการทดลองกับการตั้งค่า 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)
ObjectStore อนุญาตให้รวมเข้ากับ:
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
