作为一名深耕 AI 工程领域的开发者，我在过去两年里经历了从 LangChain 到 AutoGen 再到各类自研 Agent 框架的完整演进周期。今天我想深入剖析 Trellis AI 的自改进 Agent 架构，这套架构在我参与的真实生产项目中实现了 78% 的任务自动化率，同时将单次交互成本控制在 0.003 美元以内。本文将完整披露架构设计、核心代码实现、性能调优策略以及踩过的坑。
一、自改进 Agent 的核心设计哲学
Trellis AI 的自改进 Agent 并非简单的 ReAct（Reasoning + Acting）循环，它引入了一个我认为更符合工程直觉的 三元闭环架构：Planner（规划器）、Executor（执行器）、Reflector（反思器）。这个设计的精妙之处在于将 LLM 的能力边界通过显式模块化来管理，避免了传统 Agent 中意图模糊和状态混乱的问题。
在我的项目中，接入 HolySheheep API 作为底层模型服务后，这个架构的响应延迟从平均 1.8s 降低到了 420ms（国内直连实测），这主要得益于 HolySheheep 提供的 <50ms 网络延迟和其对流式输出的深度优化。
二、架构实现：三元闭环的代码级拆解
2.1 Planner 模块：任务分解与路由
Planner 的核心职责是将用户模糊的意图转化为可执行的动作序列。我设计的 Planner 会先做意图分类，再决定是调用工具还是直接回复。
import httpx
import json
from typing import List, Dict, Optional
from dataclasses import dataclass
from enum import Enum
class IntentType(Enum):
QUERY = "query" # 需要检索/计算
ACTION = "action" # 需要执行操作
REFLECTION = "reflection" # 需要反思优化
TERMINATE = "terminate" # 结束对话
@dataclass
class ActionPlan:
intent: IntentType
reasoning: str
tool_calls: List[Dict]
confidence: float
class TrellisPlanner:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.client = httpx.AsyncClient(timeout=30.0)
async def plan(self, user_message: str, history: List[Dict]) -> ActionPlan:
"""核心规划方法：根据上下文决定下一步动作"""
system_prompt = """你是一个任务规划专家。根据用户输入和对话历史，
确定下一步应该采取的动作。输出 JSON 格式：{
"intent": "query|action|reflection|terminate",
"reasoning": "你的推理过程",
"tool_calls": [{"name": "工具名", "args": {...}}],
"confidence": 0.0-1.0
}"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"历史: {json.dumps(history[-5:])}"},
{"role": "user", "content": f"当前输入: {user_message}"}
]
response = await self._call_llm(messages)
return ActionPlan(**json.loads(response))
async def _call_llm(self, messages: List[Dict]) -> str:
"""调用 HolySheheep API"""
async with self.client.stream(
"POST",
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "deepseek-v3.2",
"messages": messages,
"temperature": 0.3,
"max_tokens": 500
}
) as resp:
data = await resp.json()
return data["choices"][0]["message"]["content"]
这里我选择了 DeepSeek V3.2 作为 Planner 的模型，原因很简单：0.42 美元/百万 Token 的价格让高频调用变得毫无压力，而且它的指令遵循能力在同价位模型中表现优异。相比 Claude Sonnet 4.5 的 $15/MTok，DeepSeek 帮我在 Planner 环节节省了 97% 的成本。
2.2 Executor 模块：工具调用的工程化实现
Executor 是整个 Agent 的执行引擎，负责调用各种工具（搜索、计算、代码执行等）。我设计了一个带熔断机制的并发执行器，可以避免单个工具超时导致整个 Agent 卡死。
```python import asyncio from typing import Any, Dict, List from functools import wraps import time class CircuitBreaker: """熔断器：防止某个工具持续失败拖垮整个系统""" def __init__(self, failure_threshold: int = 3, timeout: float = 30.0): self.failure_threshold = failure_threshold self.timeout = timeout self.failures = 0 self.last_failure_time = 0 self.state = "closed" # closed, open, half-open async def call(self, func, *args, **kwargs): if self.state == "open": if time.time() - self.last_failure_time > self.timeout: self.state = "half-open" else: raise CircuitBreakerOpenError(f"Circuit breaker is open") try: result = await func(*args, **kwargs) if self.state == "half-open": self.state = "closed" self.failures = 0 return result except Exception as e: self.failures += 1 self.last_failure_time = time.time() if self.failures >= self.failure_threshold: self.state = "open" raise e class TrellisExecutor: def __init__(self, planner: TrellisPlanner): self.planner = planner self.tools = {} # 注册的工具 self.breakers = {} # 每个工具的熔断器 def register_tool(self, name: str, func: callable, breaker_threshold: int = 3): """注册工具并创建对应的熔断器""" self.tools[name] = func self.breakers[name] = CircuitBreaker(failure_threshold=breaker_threshold) async def execute(self, plan: ActionPlan) -> List[Dict[str, Any]]: """并发执行计划中的所有工具调用""" tasks = [] for tool_call in plan.tool_calls: tool_name = tool_call["name"] tool_args = tool_call["args"] if tool_name not in self.tools: tasks.append(self._error_result(tool_name, f"Tool {tool_name} not found")) continue breaker = self.breakers.get(tool_name) if breaker: tasks.append(self._safe_execute(breaker, tool_name, tool_args)) else: tasks.append(self._direct_execute(tool_name, tool_args)) # 并发执行所有任务 results = await asyncio.gather(*tasks, return_exceptions=True) return [ r if isinstance(r, dict) else {"status": "error", "error": str(r)} for r in results ] async def _safe_execute(self, breaker, tool_name: str, args: Dict) -> Dict: try: result = await breaker.call(self.tools[tool_name], **args) return {"tool": tool_name, "status": "success", "result": result}