Flow (Laminar): eine leichtgewichtige Task-Engine für den Aufbau von Intelligenz, die Aufgaben vereinfacht und flexibel verwaltet

Allgemeine Einführung

Flow ist eine leichtgewichtige Task-Engine, die für die Entwicklung von KI-Agenten entwickelt wurde, wobei der Schwerpunkt auf Einfachheit und Flexibilität liegt. Im Gegensatz zu traditionellen knoten- und kantenbasierten Workflows verwendet Flow ein dynamisches Aufgabenwarteschlangensystem, das parallele Ausführung, dynamische Planung und intelligentes Abhängigkeitsmanagement unterstützt. Flow benötigt keine vordefinierten Kanten zwischen Knoten und verwendet eine dynamische Task-Scheduling-Architektur, um Entwicklern zu helfen, sauberen, leicht verständlichen Code zu schreiben. Flow wird vom Laminar-Team gepflegt, unterstützt automatisches Tracing und Zustandsmanagement und ist für eine Vielzahl von KI-Anwendungsszenarien geeignet.

Funktionsliste

• Parallele Task-Ausführung: Automatische parallele Ausführung von Tasks ohne expliziten Threading-Code.
• Dynamische Planung: Aufgaben können zur Laufzeit neue Aufgaben planen.
• Intelligentes Abhängigkeitsmanagement: Aufgaben können auf das Ergebnis eines vorherigen Vorgangs warten.
• Zustandsverwaltung: Speichern und Laden der Zustände von Aufgaben, beginnend mit der Ausführung einer bestimmten Aufgabe.
• Bedingte Verzweigungen und Kontrollfluss: Bedingte Verzweigungen und Schleifensteuerung werden unterstützt.
• Streaming Task Execution: Unterstützt das Streaming der Task-Ausführung.
• Automatisiertes Tracing: Unterstützt das automatisierte Tracing von OpenTelemetry zur einfachen Fehlersuche und Zustandsrekonstruktion.
• Geringes Gewicht und keine externen Abhängigkeiten: Das Design ist einfach, flexibel und leistungsstark.

Hilfe verwenden

Einbauverfahren

Um Flow zu installieren, verwenden Sie einfach den Befehl pip:

pip install lmnr-flow

Grundlegende Verwendung

Nachfolgend finden Sie ein einfaches Beispiel für seine Verwendung:

from concurrent.futures import ThreadPoolExecutor
from lmnr_flow import Flow, TaskOutput, NextTask, Context
# 创建Flow实例
flow = Flow(thread_pool_executor=ThreadPoolExecutor(max_workers=4))
# 定义一个简单任务
def my_task(context: Context) -> TaskOutput:
return TaskOutput(output="Hello World!")
# 添加任务到Flow
flow.add_task("greet", my_task)
# 运行任务
result = flow.run("greet")
print(result)  # 输出: {"greet": "Hello World!"}

Missionskette

Quests können andere Quests auslösen:

def task1(context: Context) -> TaskOutput:
return TaskOutput(output="result1", next_tasks=[NextTask("task2")])
def task2(context: Context) -> TaskOutput:
t1_result = context.get("task1")
return TaskOutput(output="result2")
flow.add_task("task1", task1)
flow.add_task("task2", task2)
flow.run("task1")  # 输出: {"task2": "result2"}

parallele Ausführung

Mehrere Aufgaben können parallel ausgeführt werden:

def starter(context: Context) -> TaskOutput:
return TaskOutput(output="started", next_tasks=[NextTask("slow_task1"), NextTask("slow_task2")])
def slow_task1(context: Context) -> TaskOutput:
time.sleep(1)
return TaskOutput(output="result1")
def slow_task2(context: Context) -> TaskOutput:
time.sleep(1)
return TaskOutput(output="result2")
flow.add_task("starter", starter)
flow.add_task("slow_task1", slow_task1)
flow.add_task("slow_task2", slow_task2)
flow.run("starter")  # 两个任务并行执行，总耗时约1秒

Streaming-Ergebnisse

Aufgaben können gestreamt werden, um Zwischenergebnisse zu liefern:

def streaming_task(context: Context) -> TaskOutput:
stream = context.get_stream()
for i in range(3):
stream.put(StreamChunk("streaming_task", f"interim_{i}"))
return TaskOutput(output="final")
flow.add_task("streaming_task", streaming_task)
for task_id, output in flow.stream("streaming_task"):
print(f"{task_id}: {output}")

Dynamischer Arbeitsablauf

Aufgaben können auf der Grundlage von Bedingungen dynamisch geplant werden:

def conditional_task(context: Context) -> TaskOutput:
count = context.get("count", 0)
if count >= 3:
return TaskOutput(output="done")
context.set("count", count + 1)
return TaskOutput(output=f"iteration_{count}", next_tasks=[NextTask("conditional_task")])
flow.add_task("conditional_task", conditional_task)
flow.run("conditional_task")  # 任务循环3次后完成

Eingabeparameter

Aufgaben können Eingabeparameter erhalten:

def parameterized_task(context: Context) -> TaskOutput:
name = context.get("user_name")
return TaskOutput(output=f"Hello {name}!")
flow.add_task("greet", parameterized_task)
result = flow.run("greet", inputs={"user_name": "Alice"})
print(result)  # 输出: {"greet": "Hello Alice!"}

dynamisches Routing

Aufgaben können auf der Grundlage von Eingaben dynamisch weitergeleitet werden:

def router(context: Context) -> TaskOutput:
task_type = context.get("type")
routes = {
"process": [NextTask("process_task")],
"analyze": [NextTask("analyze_task")],
"report": [NextTask("report_task")]
}
return TaskOutput(output=f"routing to {task_type}", next_tasks=routes.get(task_type, []))
flow.add_task("router", router)
flow.add_task("process_task", lambda ctx: TaskOutput("processed data"))
flow.run("router", inputs={"type": "process"})  # 输出: {"process_task": "processed data"}

Statusverwaltung

Aufgabenstatus können gespeichert und geladen werden:

context = Context()
context.from_dict({"task1": "result1"})
flow = Flow(context=context)
flow.add_task("task2", lambda ctx: TaskOutput("result2"))
flow.run("task2")
assert flow.context.get("task1") == "result1"
assert flow.context.get("task2") == "result2"

Map Reduce

Tasks können Map-Reduce-Operationen durchführen:

def task1(ctx):
ctx.set("collector", [])
return TaskOutput("result1", next_tasks=[NextTask("task2", spawn_another=True) for _ in range(3)])
def task2(ctx):
collector = ctx.get("collector")
collector.append("result2")
ctx.set("collector", collector)
return TaskOutput("", next_tasks=[NextTask("task3")])
def task3(ctx):
collector = ctx.get("collector")
return TaskOutput(collector)
flow.add_task("task1", task1)
flow.add_task("task2", task2)
flow.add_task("task3", task3)
result = flow.run("task1")
assert result == {"task3": ["result2", "result2", "result2"]}

LLM-Agent

Beispiel für einen LLM-Agenten für die dynamische Werkzeugauswahl:

from typing import List
import json
def llm_agent(context: Context) -> TaskOutput:
prompt = context.get("user_input")
llm_response = {
"reasoning": "Need to search database and format results",
"tools": ["search_db", "format_results"]
}
next_tasks = [NextTask(tool) for tool in llm_response["tools"]]
return TaskOutput(output="LLM agent response", next_tasks=next_tasks)
flow.add_task("llm_agent", llm_agent)
flow.run("llm_agent", inputs={"user_input": "Find data"})