geneflow-agent 0.1.1

Agent/Crew framework with association memory, strategy genes, and test-time evolution.

GeneFlow Agent Framework

目标：提供与 CrewAI 类似的 Agent / Task / Crew 编排体验，但内置：

• 检索 + 联想（M-Flow/Cone Graph 的简化结构）
• 策略基因（Strategy Genes，高控制密度策略注入）
• 自主进化闭环（Event -> Evaluate -> Evolve Gene -> 下次执行生效）

安装

pip install -r requirements.txt
pip install -e .

环境变量

export OPENAI_API_KEY="your-key"
export OPENAI_BASE_URL="https://api.openai.com/v1"

export PGHOST="localhost"
export PGPORT="5432"

数据库要求

本框架默认使用 PostgreSQL + pgvector。

一键初始化 schema（默认 vector(3)，对应 HashEmbeddingProvider(dim=3)）：

psql -h $PGHOST -U $PGUSER -d $PGDATABASE -f schema.sql

schema 包含：

• episodes(content text, embedding vector(3), keywords text[], created_at timestamptz)
• entities(name text unique, embedding vector(3))
• episode_entities(episode_id, entity_id)
• genes(keywords text[], summary text, strategy text, avoid_signals text, embedding vector(3), ...)
• evolution_events(query, result, is_success, feedback, execution_path, ...)

如需接入 OpenAI embedding（例如 text-embedding-3-small = 1536 维）， 请把 schema.sql 中的 vector(3) 改为 vector(1536) 并使用 OpenAIEmbeddingProvider。

连接池共享

多 Agent 场景下，所有 Agent 共享同一连接池（按 db_params 归并）：

from geneflow.db import DatabaseManager
DatabaseManager.close_all()  # 退出前释放

测试时仍可 DatabaseManager(db_params) 直接构造独立池。

运行单元测试

不需要真实数据库或 LLM：

python -m unittest test_smoke -v

最小 Demo：一次失败 -> 生成 Gene -> 第二次成功

python demo_minimal.py

这个 demo 会：

1. 插入一个 Episode（作为联想记忆）
2. 第一次运行强制判定失败（Evaluator 返回失败）
3. 触发 GeneManager.evolve_gene(...) 写入新的 Gene（包含 FORCE_FINAL: 控制信号）
4. 第二次运行匹配到 Gene 后，直接输出 FORCE_FINAL: 后面的内容，从而判定成功

API 速览（CrewAI 风格）

from openai import OpenAI
from geneflow import Agent, Task, Crew

db_params = {"dbname": "vector_db", "user": "root", "password": "root", "host": "localhost", "port": 5432}
llm = OpenAI()

a1 = Agent(db_params, llm, name="researcher", role="researcher", goal="gather facts")
a2 = Agent(db_params, llm, name="writer", role="writer", goal="write final answer")

tasks = [
  Task(id="t1", description="收集要点", expected_output="要点列表", agent=a1),
  Task(id="t2", description="输出最终方案", expected_output="可执行步骤", context_from=["t1"], agent=a2),
]

crew = Crew(tasks)
outputs = crew.kickoff(inputs={"topic": "xxx"})
print(outputs["t2"])

扩展流程（Router / Hierarchical）

Router：任务不绑定 agent，由 router 选择执行者。

from geneflow import Agent, Task, Crew
from openai import OpenAI

llm = OpenAI()
db_params = {"dbname": "vector_db", "user": "root", "password": "root", "host": "localhost", "port": 5432}

router = Agent(db_params, llm, name="router", role="router", goal="route tasks to best agent")
a1 = Agent(db_params, llm, name="researcher", role="researcher", goal="gather facts")
a2 = Agent(db_params, llm, name="coder", role="engineer", goal="write code")

tasks = [
  Task(id="t1", description="找出问题原因", expected_output="原因", agent=None),
  Task(id="t2", description="给出修复方案", expected_output="步骤", context_from=["t1"], agent=None),
]

crew = Crew(tasks, process="router", agents=[a1, a2], router=router)
outputs = crew.kickoff()

Hierarchical：manager 先给出执行顺序（以及可选的 agent_name），再按计划执行。

from geneflow import Agent, Task, Crew
from openai import OpenAI

llm = OpenAI()
db_params = {"dbname": "vector_db", "user": "root", "password": "root", "host": "localhost", "port": 5432}

manager = Agent(db_params, llm, name="manager", role="manager", goal="plan task order and assign agents")
router = Agent(db_params, llm, name="router", role="router", goal="route tasks to best agent")
a1 = Agent(db_params, llm, name="researcher", role="researcher", goal="gather facts")
a2 = Agent(db_params, llm, name="writer", role="writer", goal="write final answer")

tasks = [
  Task(id="t1", description="收集要点", expected_output="要点列表", agent=None),
  Task(id="t2", description="输出最终方案", expected_output="可执行步骤", context_from=["t1"], agent=None),
]

crew = Crew(tasks, process="hierarchical", agents=[a1, a2], router=router, manager=manager)
outputs = crew.kickoff()

Hierarchical Step 级能力（指派 / 重写 / 重试）

当 process="hierarchical" 时，manager 可以在 steps 中返回以下字段：

• agent_name：强制指定由哪个 agent 执行该 step（找不到则 fallback 到 router 或第一个 agent）
• override_description：重写本次执行的 task 描述（不改原 Task 对象）
• override_expected_output：重写期望输出
• max_retries：失败自动重试次数；每次重试会把上一次的 result 与 feedback 注入到 context，并触发 Gene 约束

manager JSON 输出示例：

{
  "steps": [
    {
      "task_id": "t1",
      "agent_name": "researcher",
      "override_description": "用列表列出3个关键风险点",
      "override_expected_output": "3条风险点",
      "max_retries": 1
    },
    {
      "task_id": "t2",
      "agent_name": "writer",
      "max_retries": 2
    }
  ],
  "add_tasks": []
}

并行（Parallel）

Parallel：自动按依赖关系调度，并行执行无依赖或依赖已满足的任务。

from geneflow import Agent, Task, Crew
from openai import OpenAI

llm = OpenAI()
db_params = {"dbname": "vector_db", "user": "root", "password": "root", "host": "localhost", "port": 5432}

a1 = Agent(db_params, llm, name="a1", role="researcher", goal="collect")
a2 = Agent(db_params, llm, name="a2", role="engineer", goal="implement")

tasks = [
  Task(id="t1", description="收集方案A要点", expected_output="要点", agent=a1),
  Task(id="t2", description="收集方案B要点", expected_output="要点", agent=a1),
  Task(id="t3", description="汇总对比并输出建议", expected_output="建议", context_from=["t1", "t2"], agent=a2),
]

crew = Crew(tasks, process="parallel")
outputs = crew.kickoff()

Map-Reduce

Map-Reduce：先并行跑 map tasks，再执行 reducer task（reducer 会拿到所有 map 输出作为 context）。

from geneflow import Agent, Task, Crew
from openai import OpenAI

llm = OpenAI()
db_params = {"dbname": "vector_db", "user": "root", "password": "root", "host": "localhost", "port": 5432}

mapper = Agent(db_params, llm, name="mapper", role="researcher", goal="map")
reducer = Agent(db_params, llm, name="reducer", role="writer", goal="reduce")

tasks = [
  Task(id="m1", description="分析维度1", expected_output="结论", agent=mapper, reduce=False),
  Task(id="m2", description="分析维度2", expected_output="结论", agent=mapper, reduce=False),
  Task(id="r1", description="合并m1/m2输出给最终建议", expected_output="最终建议", agent=reducer, reduce=True),
]

crew = Crew(tasks, process="map_reduce")
outputs = crew.kickoff()
print(outputs["r1"])

Dynamic Tasks（Hierarchical 增量生成）

当 process="hierarchical" 且提供 manager 时，manager 可以返回：

• steps: 任务执行顺序
• add_tasks: 新增任务（会被注入到本次执行图里，然后按 steps 执行）

这用于实现“经理动态拆任务/补任务/重排任务”的能力。

目录结构

• geneflow/agent.py：Agent/Task/Crew/Tool + 进化闭环
• geneflow/memory.py：联想记忆（向量召回 + 实体锚点扩展）
• geneflow/genes.py：策略基因（匹配 + 演化写回）
• geneflow/embedding.py：EmbeddingProvider（Hash/OpenAI）
• geneflow/evaluator.py：Evaluator（LLM/规则/状态机）