omnigent (Francesco Stabile)

全能型

构建自主AI代理的通用支架。

在经过生产验证的基础上构建任何人工智能代理——安全、代码分析、DevOps、合规性、研究。从具有17k+LOC和320个测试的真实世界代理中提取。

   

建筑 · 例子 · 贡献

______________________________________________________________________

什么是Omnigent？

大多数AI代理框架都为您提供了LLM API的包装器。Omnigent为您提供 整个大脑.

这是生产自主代理的领域无关架构——ReAct循环、多提供商LLM路由、结构化内存、分层规划、推理图、错误恢复、反射和插件系统。你需要建立一个 真实的 代理，而不是带有工具的聊天机器人。

你带来了域名。全能者带来智慧。

┌──────────────────────────────────────────────────────┐
│                   Agent Loop (ReAct)                 │
│         Reason → Act → Observe → Reflect             │
├──────────────┬────────────┬────────────┬─────────────┤
│   Router     │  Planner   │  Context   │   Graph     │
│  4 Providers │  Phases    │  Smart Trim│  Reasoning  │
├──────────────┴────────────┴────────────┴─────────────┤
│            Post-Processing Pipeline                  │
│    Extractors → Reflection → Error Recovery          │
├──────────────────────────────────────────────────────┤
│            Tool Registry + Plugin System             │
├──────────────────────────────────────────────────────┤
│  State │ DomainProfile │ Session │ Cost │ Knowledge  │
├──────────────────────────────────────────────────────┤
│          Config │ Logging │ MCP Integration          │
└──────────────────────────────────────────────────────┘

为什么是Omnigent？

快速开始

安装

pip install -e .

设置API密钥

export DEEPSEEK_API_KEY="sk-..."  # Cheapest option (~$0.001 per analysis)
# or
export ANTHROPIC_API_KEY="sk-ant-..."
# or
export OPENAI_API_KEY="sk-..."

运行示例代理

# CodeLens — code quality analyzer (included example)
python -m examples.codelens.main /path/to/any/project

通过4个步骤构建自己的代理

步骤1：定义域内存

from dataclasses import dataclass, field
from omnigent.domain_profile import DomainProfile

@dataclass
class MyProfile(DomainProfile):
    items_analyzed: list[str] = field(default_factory=list)
    risk_score: float = 0.0

第二步：注册您的工具

from omnigent.tools import ToolRegistry

registry = ToolRegistry()
registry.register(
    name="my_scanner",
    schema={"description": "Scan a target", "parameters": {
        "type": "object",
        "properties": {"target": {"type": "string"}},
        "required": ["target"],
    }},
    handler=my_scanner_function,
)

步骤3：填充注册表 （计划模板、链、提取器、反射器、错误模式）

from omnigent.registry import DomainRegistry
from omnigent.chains import ChainStep

registry = DomainRegistry(
    plan_templates={
        "my_domain": [
            {"name": "Discovery", "objective": "Map the target", "steps": [
                ("Initial scan", "my_scanner"),
            ]},
        ],
    },
    chains={
        "high_risk": [
            ChainStep("Deep dive on flagged items", "deep_scanner"),
            ChainStep("Generate remediation plan", ""),
        ],
    },
    extractors={
        "my_scanner": lambda profile, result, args: setattr(
            profile, 'risk_score', 0.8
        ),
    },
)

第四步：将其连接并运行

import asyncio
from omnigent.agent import Agent
from omnigent.router import LLMRouter, Provider

async def main():
    agent = Agent(
        router=LLMRouter(primary=Provider.DEEPSEEK),
        tools=tool_registry,
        registry=registry,  # DomainRegistry with all domain-specific behavior
    )
    async for event in agent.run("Analyze this target"):
        if event.type == "text":
            print(event.content, end="")
        elif event.type == "finding":
            print(f"\n[{event.finding.severity}] {event.finding.title}")

asyncio.run(main())

你能建造什么？

Omnigent是 领域无关 --它提供了智能架构，你提供了领域知识：

看 示例/代码镜头/ 以实现完整的工作实施。

组件

关键设计模式

数据驱动注册表（核心零域代码）

所有领域特定的行为都存在于一个注射中 DomainRegistry 数据类。您的代理在启动时填充它：

from omnigent.registry import DomainRegistry

registry = DomainRegistry(
    plan_templates={...},   # Task plan templates
    chains={...},           # Escalation chains
    extractors={...},       # Tool result parsers
    reflectors={...},       # Post-tool strategic analysis
    error_patterns={...},   # Failure recovery patterns
    knowledge_map={...},    # Knowledge file routing
    examples={...},         # Few-shot tool examples
    tool_timeouts={...},    # Per-tool timeouts
)

agent = Agent(registry=registry)

多个代理可以使用独立的注册表运行——没有全局状态泄漏。为了向后兼容性， DomainRegistry.default() 读取模块级字典。

复杂域的子类

对于不能表示为数据的行为，重写方法：

class MyAgent(Agent):
    # Domain hooks
    def _is_failure(self, tool_name, result): ...
    def _extract_finding(self, text): ...
    def _build_dynamic_system_prompt(self): ...

    # Overridable step methods (decomposed agent loop)
    def _do_context_management(self): ...
    async def _do_llm_call(self, system_prompt): ...
    async def _do_tool_execution(self, tool_calls): ...
    async def _do_post_tool_processing(self, tc, result): ...
    def _check_termination(self, text_buffer): ...

class MyGraph(ReasoningGraph):
    def _build_default_graph(self): ...

class MyProfile(DomainProfile):
    def to_prompt_summary(self): ...

推理图（微分器）

Omnigent与简单工具调用者的区别在于：当发现得到确认时，推理图会激活下游升级路径。代理人不仅发现问题 将它们束缚在多步骤推理中.

Security:    SQLi → DB Dump → Credential Extraction → Admin Access → RCE
Code:        God Object → High Coupling → Low Testability → Regression Risk
Incident:    Alert → Log Correlation → Root Cause → Blast Radius
Compliance:  Gap Found → Control Missing → Risk Assessment → Remediation

测试

# Run all tests
pytest

# Run with coverage
pytest --cov=omnigent --cov-report=term-missing

# Run specific module tests
pytest tests/test_reasoning_graph.py -v

# Run only unit tests
pytest -m unit

325项测试涵盖所有核心组件。每个测试都在没有LLM调用或网络访问的情况下运行。

项目结构

omnigent/
├── src/omnigent/         # Core framework
│   ├── agent.py           # The ReAct loop (1024 lines)
│   ├── registry.py        # DomainRegistry dataclass (97 lines)
│   ├── router.py          # Multi-provider LLM routing + LLMProvider ABC (700 lines)
│   ├── reasoning_graph.py # Chain reasoning engine (389 lines)
│   ├── planner.py         # Hierarchical task planner + macro-reflection (544 lines)
│   ├── context.py         # Smart context + semantic compression (358 lines)
│   ├── state.py           # State + Pydantic findings
│   ├── domain_profile.py  # Structured agent memory (bounded summaries)
│   ├── extractors.py      # Result parsing pipeline
│   ├── reflection.py      # Async post-tool strategic analysis
│   ├── error_recovery.py  # Failure recovery engine
│   ├── chains.py          # Escalation chain registry
│   ├── knowledge_loader.py # Knowledge base retrieval
│   ├── few_shot_examples.py # Tool usage examples
│   ├── plugins.py         # Plugin system + strict checksum mode (556 lines)
│   ├── session.py         # Session persistence + checkpoint/replay (546 lines)
│   ├── cost_tracker.py    # Cost tracking
│   ├── config.py          # Configuration management
│   ├── logging_config.py  # Structured JSON logging
│   ├── prompts/system.md  # Base system prompt
│   └── tools/             # Tool registry + schema caching (309 lines)
├── examples/codelens/     # Complete working example agent
├── tests/                 # 325 tests
├── ARCHITECTURE.md        # Deep technical architecture doc
├── CONTRIBUTING.md        # Contribution guide
└── CHANGELOG.md           # Version history

起源

Omnigent提取自 NumaSec，生产自主安全代理（17878 LOC，320测试）。

由...建造 弗朗切斯科·斯塔比尔

 

许可证

麻省理工学院 --把它用于任何事情。

贡献

看 贡献.md。我们欢迎域实现、错误修复和文档改进。