Agent Orchestrator Layer

代理编排器层

围绕目标的多域系统，通过LLM进行意图提取，元数据驱动的规划和确定性DAG执行。

架构概述

User Input
  ↓ EntryRequest
Intent Adapter (LLM)
  ↓ IntentOutput { primary_domain, goal, entities{*_text / enum} }
Goal Resolver (deterministic)
  ↓ ExecutionIntent { domain, capability, parameters, confidence }
Planner Service + Memory
  ↓ ExecutionPlan { steps[], execution_mode, combine_mode }
Execution Engine (DAG)
  ↓ ExecutionIntent (per step)
Orchestrator (registry lookup)
  ↓
Domain Handler
  ↓ DomainOutput { status, result, explanation }

每层和有效载荷的完整细节： 建筑.md

______________________________________________________________________

图表

graph TD
    User((User)) --> Entry[Entry Layer\nCLI / Telegram / HTTP]
    Entry --> Intent[Intent Adapter\nLLM → IntentOutput]
    Intent --> Resolver[Goal Resolver\ngoal → capability]
    Resolver --> Planner[Planner Service\n+ Memory injection]
    Mem[(Memory Store)] --> Planner
    Planner --> Decomposer[TaskDecomposer\nmetadata-driven]
    Planner --> FCPlanner[FunctionCallingPlanner\noptional LLM loop]
    Decomposer --> Exec[Execution Engine\nDAG / parallel]
    FCPlanner --> Exec
    Exec --> Orch[Orchestrator\nconfidence gate + routing]
    Orch --> Reg[Registry]
    Reg --> Fin[Finance Domain\nremote_http :8001]
    Reg --> Com[Communication Domain\nremote_http :8002]

______________________________________________________________________

每层有效载荷

入场申请

EntryRequest(
    session_id="user-abc123",
    input_text="qual o preço da Nordea?",
    metadata={}
)

IntentOutput（Intent Adapter的输出）

LLM提取 目标 和 对人类友好的实体 --从不使用技术性股票或ID。

IntentOutput(
    primary_domain="finance",
    goal="GET_QUOTE",
    entities={"symbol_text": "Nordea"},   # name as the user said it
    confidence=0.95,
    original_query="qual o preço da Nordea?"
)

# Goal with enum (TOP_MOVERS)
IntentOutput(
    primary_domain="finance",
    goal="TOP_MOVERS",
    entities={"direction": "GAINERS", "market_text": "Brasil"},
    confidence=0.92,
    original_query="maiores altas do Brasil"
)

ExecutionIntent（目标解析器的输出）

确定性映射 goal + entities → capability没有法学硕士。

ExecutionIntent(
    domain="finance",
    capability="get_stock_price",          # resolved by GoalResolver
    parameters={"symbol_text": "Nordea"},  # entities become parameters
    confidence=0.95,
    original_query="qual o preço da Nordea?"
)

执行计划（Planner的输出）

ExecutionPlan(
    execution_mode="dag",
    combine_mode="report",
    steps=[
        ExecutionStep(id=1, capability="get_stock_price",
                      params={"symbol_text": "Nordea"}, depends_on=[]),
        ExecutionStep(id=2, capability="send_telegram_message",
                      params={"message": "${1.explanation}"}, depends_on=[1], required=False)
    ]
)

DomainOutput（域处理程序的输出）

DomainOutput(
    status="success",        # "success" | "failure" | "clarification"
    result={
        "symbol": "NDA-SE.ST",
        "price": 112.5,
        "currency": "SEK",
        "_market_context": {"market": "SE", "exchange": "OMX"}
    },
    explanation="Nordea está em 112.50 SEK",
    confidence=1.0,
    metadata={}
)

______________________________________________________________________

项目结构

AgentOrchestratorLayer/
├── main.py                          # Entry CLI + Telegram
├── api/openai_server.py             # OpenAI-compatible API (Open WebUI)
│
├── intent/adapter.py                # LLM → IntentOutput
│
├── planner/
│   ├── goal_resolver.py             # IntentOutput → ExecutionIntent (deterministic)
│   ├── service.py                   # orchestrates planner + memory
│   ├── task_decomposer.py           # metadata-driven step decomposition
│   └── function_calling_planner.py  # optional LLM loop
│
├── execution/
│   ├── engine.py                    # DAG executor + workflow runtime
│   ├── result_combiner.py           # combines step outputs
│   └── task_state_store.py          # persists TaskInstance + WorkflowEvent
│
├── orchestrator/orchestrator.py     # confidence gate + capability routing
│
├── registry/
│   ├── db.py                        # SQLite: domains, capabilities, goals
│   ├── loader.py                    # loads manifests → registry
│   ├── domain_registry.py           # in-memory HandlerRegistry
│   └── http_handler.py              # handler for remote_http domains
│
├── shared/
│   ├── models.py                    # all Pydantic models
│   └── workflow_contracts.py        # MethodSpec, WorkflowSpec, TaskInstance
│
├── memory/store.py                  # SQLiteMemoryStore
├── models/selector.py               # ModelSelector (Ollama/OpenAI-compat)
├── skills/                          # SkillGateway + MCP adapter
│
├── domains/
│   ├── finance/                     # Finance domain (see domains/finance/README.md)
│   └── general/handler.py           # General domain (chat)
│
├── communication-domain/            # Communication domain (see communication-domain/README.md)
│
├── scripts/                         # test and evaluation scripts
├── domains.bootstrap.json           # domain bootstrap configuration
└── docker-compose.yml

______________________________________________________________________

领域

每个域都有自己的README，其中包含清单、功能和示例：

• 金融领域 --报价、热门人物、筛选者、历史
• 通信域 --通过Telegram发送

______________________________________________________________________

主要特点

• 基于目标的意图： LLM提取目标+人性化实体；GoalResolver映射到没有LLM的功能
• 元数据驱动分解： 分解为并行步骤是在清单中配置的，而不是在代码中配置的
• DAG执行： 具有显式依赖关系的步骤，并行执行 max_concurrency
• 声明性工作流程： MethodSpec + WorkflowSpec 对于流量 human_gate, decision, validate, call, return
• 暂停/恢复： TaskInstance 持续状态； resume_task(ClarificationAnswer) 从中断的地方继续
• 内存注入： 在分解之前将结构化内存（SQLite）注入规划器
• 符号解析器： 财务处理人员解析姓名→ 通过别名元数据获取股票信息+ search_symbol 作为后备
• 软确认： 意图与 confidence < 0.94 执行前返回澄清
• 流动： 具有增量状态更新的SSE；普通聊天的真实令牌流快速路径
• 兼容OpenAI的API： 与Open WebUI直接集成

______________________________________________________________________

配置

主要变量

# LLM / Models
OLLAMA_URL=http://localhost:11434

# Remote domains
BOOTSTRAP_DOMAINS_FILE=domains.bootstrap.json

# Databases
DB_PATH=agent.db
REGISTRY_DB_PATH=registry.db
MEMORY_DB_PATH=memory.db

# Confidence
SOFT_CONFIRM_THRESHOLD=0.94

# Telegram entry
TELEGRAM_BOT_TOKEN=...
TELEGRAM_DEFAULT_CHAT_ID=...

# OpenAI API
OPENAI_API_DEBUG_TRACE=false

域名引导(domains.bootstrap.json)

[
  {
    "name": "finance",
    "type": "remote_http",
    "config": {"url": "http://finance-server:8001", "timeout": 90.0},
    "sync_capabilities": true
  },
  {
    "name": "communication",
    "type": "remote_http",
    "config": {"url": "http://communication-domain:8002", "timeout": 15.0},
    "sync_capabilities": true
  }
]

______________________________________________________________________

如何跑步

Docker Compose

docker compose up --build

服务：

• finance-server → host :8003
• communication-domain → host :8002
• agent-api → host :8010
• open-webui → host :3000

命令行界面

python3 main.py run
python3 main.py run-telegram

API

uvicorn api.openai_server:app --host 0.0.0.0 --port 8010

终点：

• GET /health
• GET /v1/models
• POST /v1/chat/completions （与 stream: true 苏格兰和南方能源公司）

管理员

python3 main.py domain-list
python3 main.py domain-add finance remote_http '{"url":"http://localhost:8003"}'
python3 main.py domain-sync finance
python3 main.py memory-set preferred_market '"SE"'
python3 main.py memory-get preferred_market

______________________________________________________________________

测试

# unit tests
python3 -m pytest -q

# integration scripts
PYTHONPATH=. python3 scripts/test_stock_price_notify_simple.py
PYTHONPATH=. python3 scripts/test_telegram_send_simple.py

# capability evaluation (requires domains running)
FINANCE_DOMAIN_URL=http://localhost:8003 python3 scripts/evaluate_capabilities.py

______________________________________________________________________

故障排除

许可证

麻省理工学院