1. 项目概述：当AI不再只是“回答问题”，而是开始“主动做事”

你有没有试过让一个大语言模型帮你订机票？不是让它写一段描述“如何订机票”的文字，而是真的打开浏览器、填表、选航班、完成支付——当然，目前它还做不到这一步。但如果你把任务拆解成“查天气”“比价”“生成行程单”“发邮件确认”几个小动作，再给每个动作配上专用工具和明确指令，最后由一个“指挥官”来判断下一步该调用谁、什么时候调用、失败了怎么兜底……那你就已经站在了Agentic Framework（智能体框架）的实操门口。

这不是科幻设定，而是2025年工程落地的真实切口。我从去年下半年开始系统性地在三个客户项目里落地这类架构：一个是跨境电商客服后台的自动工单分派与初步响应系统，一个是本地律所的合同关键条款比对+风险提示流水线，还有一个是制造业设备维保知识库的动态问答引擎。它们共同的特点是—— 单靠一个LLM prompt无法稳定交付结果，必须靠结构化调度、状态感知、工具编排和失败重试机制来兜底 。所谓“LLM Orchestrator”，说白了就是这个指挥官：它不直接生成最终答案，而是决定“谁干、怎么干、干到哪一步、干砸了怎么办”。

关键词里的“Towards AI - Medium”只是原始出处标记，实际内容完全脱离平台语境。我们今天聊的，是 一套可复现、可调试、可监控、能跑在普通云服务器上的Python实现方案 。它不依赖任何黑盒SaaS服务，所有组件都开源可控；它不追求“最先进”，而追求“第一次跑通不报错、第二次改配置不崩溃、第三次加功能不重构”。适合两类人：一是想摆脱“prompt工程师”身份、往AI系统架构方向发展的开发者；二是技术负责人，需要评估这类框架是否真能嵌入现有业务流程，而不是又一个PPT级概念。

我不会从“什么是Agent”这种教科书定义讲起。咱们直接看一个真实场景：某市政务热线每天收到3000+条市民留言，其中60%属于“咨询类”（如“社保卡丢了怎么补办？”），30%是“投诉类”（如“XX路井盖破损无人处理”），10%是“建议类”（如“希望增加社区老年食堂”）。过去靠人工分类+转派，平均响应时间48小时。现在用我们搭的Orchestrator，15秒内完成意图识别、信息抽取、责任部门匹配、自动生成初稿回复，并推送到对应科室工作台——整个过程无人工干预，准确率92.7%，且所有决策路径可追溯、可回放、可人工覆盖。

这就是Agentic Framework的落点：它解决的从来不是“模型能不能答对”，而是“系统能不能稳稳地把事办成”。

2. 整体设计思路：为什么不用LangChain/LlamaIndex，而选择手撸核心调度层

2.1 三层解耦架构：Orchestrator ≠ Agent，更不是LLM Wrapper

很多初学者一上来就猛扎进LangChain的Chain、Agent、Tool概念里，结果两周后发现：本地跑通的demo，一上生产环境就内存爆满；调试时print出来的中间变量，和实际执行时走的分支完全对不上；换了个新模型，整个workflow就崩，连错误日志都看不懂。问题出在哪？根本原因在于—— 他们把Orchestrator、Agent、Execution Engine混为一谈了 。

我画过不下二十张架构草图，最终锁定这个三层解耦模型：

• 顶层：Orchestrator（调度中枢）
  它只做三件事：接收用户输入→解析当前全局状态（包括历史动作、已获取数据、超时计数、权限上下文）→基于预设策略（规则引擎 or 轻量决策树）选择下一个要激活的Agent模块。它本身不调用任何LLM，不处理任何文本，甚至不碰网络IO。它的输出永远是一个结构化指令： {"agent_id": "doc_parser", "input": {"file_id": "xxx", "page_range": [1,3]}} 。

• 中层：Agent（能力单元）
  每个Agent是一个独立Python类，封装一个明确职责： DocParserAgent 负责PDF文本提取与结构化； GeoLocatorAgent 负责地址标准化与经纬度查询； PolicyCheckerAgent 负责比对本地政策库并标红冲突条款。关键约束：每个Agent必须实现 can_handle(input: dict) → bool 接口，用于Orchestrator做前置路由判断；必须定义 max_retries=2 、 timeout_sec=15 等硬性SLA参数；所有外部依赖（如OCR API、数据库连接）必须通过构造函数注入，禁止硬编码。

• 底层：Execution Engine（执行沙盒）
  这才是真正干活的地方。它接收Orchestrator下发的指令，启动对应Agent实例，在独立进程/线程/容器中执行，并强制施加资源限制（CPU 20%、内存300MB、超时kill）。执行结果无论成功失败，都统一包装成 ExecutionResult(status: str, output: dict, error: str, metrics: dict) 返回给Orchestrator。这里没有魔法——只有 subprocess.run() 、 threading.Timer 、 psutil.Process().memory_info() 这些老朋友。

提示：这种解耦带来的最大好处是测试友好性。Orchestrator单元测试只需mock Agent的 can_handle() 和 execute() 方法，完全不依赖真实LLM；Agent测试可离线运行，用预存的PDF样本和mock的OCR返回值；Execution Engine测试则专注资源控制逻辑。上线前，我们能保证每一层都100%覆盖。

2.2 为什么放弃LangChain的AgentExecutor？

LangChain的 AgentExecutor 确实省事，但它的“省事”是建立在牺牲可控性上的。我列几个真实踩坑案例：

• 状态丢失黑洞 ：当一个Agent调用外部API失败重试3次后， AgentExecutor 会把整个 intermediate_steps 列表塞进下一轮LLM的prompt。结果就是——第5轮prompt里混着前4轮的乱码日志、半截JSON、超时错误堆栈，LLM直接胡言乱语。而我们的Execution Engine要求：每次执行必须携带干净的 input 字典，失败时只返回 error 字段，绝不污染上下文。

• 工具调用不可审计 ： AgentExecutor 的 tool 注册机制是全局单例。你在A项目里注册了 search_web 工具，B项目里不小心import了A的模块， search_web 就自动出现在B的可用工具列表里——线上环境出现过因工具名冲突导致工单被错误推送给销售部的事故。

• 超时控制形同虚设 ： AgentExecutor 的 max_iterations=15 参数，指的是LLM调用次数，不是真实耗时。我们遇到过一次LLM响应延迟120秒，但 AgentExecutor 还在傻等第16次调用，导致整个请求卡死。

所以我的选择很干脆： 用LangChain的 ChatPromptTemplate 做prompt管理，用LlamaIndex的 VectorStore 做知识检索，但Orchestrator核心调度层，全部手写 。代码量其实不大——主调度循环不到200行，状态机定义80行，Execution Engine封装150行。换来的是：每毫秒的执行耗时可监控、每次工具调用可审计、每个失败节点可精准回滚。

2.3 LLM选型：不是越大越好，而是越“听话”越好

很多人以为Orchestrator必须配GPT-4或Claude-3这种顶级模型。实测下来，这是最大的误区。在我们的三个生产项目中，Orchestrator层LLM全部使用 Qwen2-7B-Instruct（量化版，显存占用<5GB） ，原因很实在：

• 指令遵循稳定性高 ：Qwen2对 <|im_start|>system\n你是一个严格按JSON Schema输出的调度器...<|im_end|> 这类强约束prompt，服从率99.2%，远超GPT-3.5的87%（我们用1000条测试用例统计）。

• 推理延迟低 ：在T4 GPU上，Qwen2-7B平均首token延迟120ms，GPT-3.5 Turbo API平均380ms（含网络往返）。对于需要高频决策的Orchestrator，这直接决定吞吐量上限。

• 成本可控 ：自托管Qwen2-7B，单卡日均电费+运维成本≈12元；调用GPT-4 API，同等QPS下月账单超2万元。

当然，Agent层可以按需混用模型： DocParserAgent 用Qwen2-7B足够， PolicyCheckerAgent 需要强法律逻辑，则切换至DeepSeek-Coder-32B（专精代码/规则解析）。Orchestrator只管调度，不管模型——这才是真正的解耦。

3. 核心细节解析：Orchestrator状态机与Agent生命周期管理

3.1 状态机不是“有限状态”，而是“带记忆的决策流”

传统FSM（有限状态机）有明确的状态集合和转移条件，比如“空闲→接收请求→解析意图→调用Agent→等待结果→返回响应”。但Agentic场景下，状态是动态生成的。举个例子：当Orchestrator处理一条“我要投诉井盖破损”的工单时，它可能触发以下分支：

1. 先调用 GeoLocatorAgent 定位地址 → 成功则进入“地址已确认”状态，失败则跳转“人工介入”状态；
2. 地址确认后，调用 DeptMatcherAgent 匹配责任部门 → 若匹配到“市政工程处”，则进入“部门已分配”状态；若匹配到“街道办”，则需额外调用 PolicyCheckerAgent 判断是否属地管理权限；
3. 所有Agent返回后，Orchestrator需综合 GeoLocator 的坐标精度、 DeptMatcher 的置信度、 PolicyChecker 的风险等级，生成最终决策：是直派工单，还是升级至区级平台。

这个过程无法用预设的5个状态穷举。我们的解法是： 用JSON Schema定义“状态快照”（State Snapshot） ，每次Agent执行完毕，Orchestrator将关键字段存入快照：

{
  "session_id": "sess_abc123",
  "current_step": "dept_matching",
  "history": [
    {
      "agent_id": "geo_locator",
      "input": {"address": "XX路与YY街交叉口"},
      "output": {"lat": 31.234, "lng": 121.456, "accuracy": "high"},
      "timestamp": "2025-04-15T10:22:33Z"
    }
  ],
  "context": {
    "user_intent": "complaint",
    "urgency_level": "high",
    "jurisdiction": "district_x"
  }
}

Orchestrator的决策函数 next_agent(state: dict) → Optional[AgentInvocation] ，接收的就是这个动态快照。它内部是一个轻量规则引擎：

def next_agent(state):
    if state["current_step"] == "init":
        return AgentInvocation("geo_locator", state["input"])
    
    # 地址定位完成后，检查精度
    last_geo = state["history"][-1]
    if last_geo["agent_id"] == "geo_locator" and last_geo["output"]["accuracy"] == "low":
        return AgentInvocation("human_fallback", {"reason": "low_accuracy_geo"})
    
    # 精度达标，进入部门匹配
    if not state.get("dept_matched"):
        return AgentInvocation("dept_matcher", {"coords": last_geo["output"]})
    
    # 部门匹配完成，生成工单
    return AgentInvocation("ticket_generator", state)

注意： AgentInvocation 是一个纯数据类，只包含 agent_id 、 input 、 timeout 三个字段。它不包含任何业务逻辑，确保Orchestrator绝对轻量。

3.2 Agent的“契约式开发”：每个Agent必须签三份协议

为了让Agent真正成为可插拔的模块，我们强制所有Agent开发者签署三份“契约”（实际是代码约束）：

第一份：输入契约（Input Contract）
每个Agent的 execute() 方法必须接受且仅接受一个 dict 类型参数，且该 dict 的key必须在类属性 INPUT_SCHEMA 中明确定义。例如：

class DocParserAgent(Agent):
    INPUT_SCHEMA = {
        "file_id": "str",      # 必填，文档唯一标识
        "page_range": "list",  # 可选，[start, end]页码范围
        "ocr_enabled": "bool"  # 可选，默认False
    }
    
    def execute(self, input_dict: dict) -> ExecutionResult:
        # 自动校验input_dict是否符合INPUT_SCHEMA
        validator = InputValidator(self.INPUT_SCHEMA)
        if not validator.validate(input_dict):
            return ExecutionResult.error(f"Invalid input: {validator.errors}")
        # ... 实际执行逻辑

第二份：输出契约（Output Contract）
Agent必须返回标准 ExecutionResult ，且 output 字段必须是 dict ，其结构由 OUTPUT_SCHEMA 声明。Orchestrator后续调用其他Agent时，会自动将前序Agent的 output 作为 input 传入，因此Schema必须兼容。

class GeoLocatorAgent(Agent):
    OUTPUT_SCHEMA = {
        "lat": "float",     # 纬度
        "lng": "float",     # 经度
        "accuracy": "str",  # "high"/"medium"/"low"
        "address_std": "str" # 标准化地址
    }

第三份：SLA契约（Service Level Agreement）
每个Agent必须声明自己的性能边界：

class PolicyCheckerAgent(Agent):
    MAX_RETRIES = 3
    TIMEOUT_SEC = 25
    MEMORY_LIMIT_MB = 512
    CPU_LIMIT_PERCENT = 30
    
    def execute(self, input_dict: dict) -> ExecutionResult:
        # Execution Engine会强制应用这些限制
        pass

这套契约机制带来两个直接收益：一是新人加入项目时，看懂 INPUT_SCHEMA 和 OUTPUT_SCHEMA 就能立刻上手写Agent，无需读全栈代码；二是Orchestrator可以基于 MAX_RETRIES 和 TIMEOUT_SEC 自动构建重试策略和熔断机制，无需每个Agent自己实现。

3.3 失败处理：不是“重试”，而是“降级+兜底+记录”

Agentic系统最怕的不是单点失败，而是失败引发的雪崩。我们的失败处理哲学是： 每一次失败都是一次明确的决策点，而非无脑重试 。

以 GeoLocatorAgent 为例，它可能失败的三种场景及应对：

失败类型	触发条件	Orchestration决策	执行动作
网络超时	requests.post() 等待>25秒	启动 GeoLocatorFallbackAgent	调用高德地图开放平台（备用API Key）
地址模糊	返回 accuracy="low" 且 confidence<0.6	跳转 HumanReviewAgent	将原始地址+模糊坐标推送到人工审核队列，标记“需确认”
服务不可用	连续3次HTTP 503	切换至 StaticGeoDBAgent	从本地SQLite地理编码库中查近似区域（精度降级为“区级”）

关键点在于： 所有兜底Agent都必须存在，且必须注册到Orchestrator的fallback registry中 。我们不允许“如果A失败就报错”这种裸奔逻辑。Orchestrator初始化时会扫描所有Agent类，自动收集 @fallback_for("geo_locator") 装饰器标注的兜底关系。

实操心得：在政务热线项目上线前，我们故意拔掉主地理编码服务的网线，观察系统行为。结果发现 StaticGeoDBAgent 返回的区级坐标，被 DeptMatcherAgent 误判为“市级部门管辖”，导致工单派错。于是我们在 StaticGeoDBAgent 的 OUTPUT_SCHEMA 里强制增加 "granularity": "district" 字段，并在 DeptMatcherAgent 的 can_handle() 中增加校验： if input.get("granularity") == "district": return False 。这种“故障驱动开发”（FDD）模式，比任何测试用例都管用。

4. 实操过程：从零搭建一个可运行的Orchestrator原型

4.1 环境准备与依赖管理：拒绝“pip install everything”

我们坚持“最小依赖原则”。整个Orchestrator核心（不含Agent实现）仅需以下7个PyPI包：

# requirements-core.txt
pydantic>=2.5.0      # 数据验证与Schema定义
tenacity>=8.2.0      # 重试策略（比retrying更轻量）
psutil>=5.9.0        # 进程资源监控
jinja2>=3.1.0        # Prompt模板渲染（比LangChain的template更可控）
python-dotenv>=1.0.0 # 环境变量管理
redis>=4.6.0         # 分布式状态存储（可选，单机用内存dict）
watchdog>=3.0.0      # Agent热加载监听（开发期用）

注意： 不安装langchain、llama-index、transformers 。这些是Agent层的可选依赖，Orchestrator核心必须保持“模型无关”。Agent开发者按需安装，比如 DocParserAgent 需要 pypdf 和 unstructured ， PolicyCheckerAgent 需要 llama_cpp_python ，但Orchestrator完全不知情。

虚拟环境创建命令（推荐conda，避免pip冲突）：

conda create -n agentic-py310 python=3.10
conda activate agentic-py310
pip install -r requirements-core.txt
# 创建基础目录结构
mkdir -p src/{orchestrator,agents,tests,examples}
touch src/__init__.py src/orchestrator/__init__.py src/agents/__init__.py

4.2 编写Orchestrator核心：200行搞定调度中枢

src/orchestrator/core.py 是心脏，代码如下（已删减注释，保留主干）：

from typing import Dict, Any, Optional, List
from pydantic import BaseModel
from tenacity import retry, stop_after_attempt, wait_exponential
import json
import time
import logging

logger = logging.getLogger(__name__)

class AgentInvocation(BaseModel):
    agent_id: str
    input: Dict[str, Any]
    timeout_sec: int = 30

class ExecutionResult(BaseModel):
    status: str  # "success" | "failed" | "timeout"
    output: Dict[str, Any] = {}
    error: str = ""
    metrics: Dict[str, Any] = {}

class Orchestrator:
    def __init__(self, agents: Dict[str, 'Agent']):
        self.agents = agents
        self.state_history: List[Dict] = []
    
    def run(self, initial_input: Dict[str, Any]) -> ExecutionResult:
        state = self._init_state(initial_input)
        while True:
            # 1. 基于当前state决定下一个Agent
            invocation = self._next_agent(state)
            if invocation is None:
                # 无更多动作，返回最终结果
                return ExecutionResult(status="success", output=state.get("final_output", {}))
            
            # 2. 执行Agent（在Execution Engine中）
            result = self._execute_agent(invocation)
            
            # 3. 更新state
            state = self._update_state(state, invocation, result)
            
            # 4. 检查终止条件（如达到最大步数、超时、人工介入）
            if self._should_terminate(state):
                break
        
        return ExecutionResult(status="terminated", output=state)
    
    def _init_state(self, input_dict: Dict[str, Any]) -> Dict[str, Any]:
        return {
            "session_id": f"sess_{int(time.time())}",
            "start_time": time.time(),
            "history": [],
            "context": input_dict,
            "step_count": 0
        }
    
    def _next_agent(self, state: Dict[str, Any]) -> Optional[AgentInvocation]:
        # 这里是你的决策逻辑，可替换为规则引擎或轻量ML模型
        step = state["step_count"]
        if step == 0:
            return AgentInvocation(agent_id="geo_locator", input=state["context"])
        elif step == 1 and state["history"][-1]["status"] == "success":
            return AgentInvocation(agent_id="dept_matcher", 
                                input=state["history"][-1]["output"])
        else:
            return None
    
    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
    def _execute_agent(self, invocation: AgentInvocation) -> ExecutionResult:
        agent = self.agents.get(invocation.agent_id)
        if not agent:
            return ExecutionResult.error(f"Agent {invocation.agent_id} not found")
        
        try:
            # Execution Engine实际调用在此处（简化版）
            result = agent.execute(invocation.input)
            return result
        except Exception as e:
            logger.error(f"Agent {invocation.agent_id} execution failed: {e}")
            return ExecutionResult.error(str(e))
    
    def _update_state(self, state: Dict[str, Any], 
                     invocation: AgentInvocation, 
                     result: ExecutionResult) -> Dict[str, Any]:
        state["history"].append({
            "agent_id": invocation.agent_id,
            "input": invocation.input,
            "output": result.output,
            "error": result.error,
            "status": result.status,
            "timestamp": time.time()
        })
        state["step_count"] += 1
        return state
    
    def _should_terminate(self, state: Dict[str, Any]) -> bool:
        if state["step_count"] > 10:
            return True
        if time.time() - state["start_time"] > 60:
            return True
        return False

这段代码的关键在于： 它不关心Agent内部怎么实现，只关心“调用-返回-更新”这个铁律 。你可以把 _next_agent() 换成一个调用小型决策树模型的函数，也可以换成读取Redis里动态配置的规则，Orchestrator核心完全不受影响。

4.3 实现第一个Agent：GeoLocatorAgent（地理定位）

src/agents/geo_locator.py ：

import requests
import json
from src.orchestrator.core import Agent, ExecutionResult
from pydantic import BaseModel, Field

class GeoLocatorInput(BaseModel):
    address: str = Field(..., description="待定位的中文地址")

class GeoLocatorOutput(BaseModel):
    lat: float = Field(..., description="纬度")
    lng: float = Field(..., description="经度")
    accuracy: str = Field(..., description="定位精度：high/medium/low")
    address_std: str = Field(..., description="标准化地址")

class GeoLocatorAgent(Agent):
    NAME = "geo_locator"
    INPUT_SCHEMA = {"address": "str"}
    OUTPUT_SCHEMA = {
        "lat": "float",
        "lng": "float", 
        "accuracy": "str",
        "address_std": "str"
    }
    MAX_RETRIES = 2
    TIMEOUT_SEC = 15
    
    def __init__(self, api_key: str):
        self.api_key = api_key
    
    def can_handle(self, input_dict: dict) -> bool:
        return "address" in input_dict and isinstance(input_dict["address"], str)
    
    def execute(self, input_dict: dict) -> ExecutionResult:
        try:
            # 调用高德地图API（示例，实际需替换为你的服务）
            params = {
                "key": self.api_key,
                "address": input_dict["address"],
                "city": "上海"
            }
            resp = requests.get(
                "https://restapi.amap.com/v3/geocode/geo",
                params=params,
                timeout=self.TIMEOUT_SEC
            )
            resp.raise_for_status()
            data = resp.json()
            
            if data["status"] != "1" or not data["geocodes"]:
                return ExecutionResult.error(f"AMap API error: {data.get('info', 'unknown')}")
            
            geocode = data["geocodes"][0]
            # 简单精度判断（实际应更复杂）
            level = geocode.get("level", "")
            accuracy = "high" if level in ["兴趣点", "门牌号"] else "medium"
            
            return ExecutionResult.success({
                "lat": float(geocode["location"].split(",")[1]),
                "lng": float(geocode["location"].split(",")[0]),
                "accuracy": accuracy,
                "address_std": geocode["formatted_address"]
            })
            
        except requests.Timeout:
            return ExecutionResult.error("Geolocation request timeout")
        except Exception as e:
            return ExecutionResult.error(f"Geolocation failed: {str(e)}")

注意 can_handle() 方法——这是Orchestrator做路由判断的依据。在真实项目中，我们会在这里加入NLP模型做地址实体识别，但最小可行版，字符串存在性检查就够了。

4.4 启动与测试：三步验证你的Orchestrator

第一步：注册Agent并启动Orchestrator

src/main.py ：

from src.orchestrator.core import Orchestrator
from src.agents.geo_locator import GeoLocatorAgent

# 初始化Agent（注入依赖）
geo_agent = GeoLocatorAgent(api_key="your_amap_key_here")

# 构建Agent注册表
agents = {
    "geo_locator": geo_agent
}

# 创建Orchestrator
orchestrator = Orchestrator(agents=agents)

# 运行一次
result = orchestrator.run({"address": "上海市浦东新区张江路123号"})
print(json.dumps(result.model_dump(), indent=2, ensure_ascii=False))

第二步：添加简单监控（5分钟上线）

在 Orchestrator.run() 开头加入：

def run(self, initial_input: Dict[str, Any]) -> ExecutionResult:
    start_time = time.time()
    logger.info(f"Orchestrator session started: {initial_input}")
    
    # ... 原有逻辑
    
    duration = time.time() - start_time
    logger.info(f"Session completed in {duration:.2f}s, steps: {state['step_count']}")
    return final_result

启动时加日志配置：

import logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

第三步：用curl快速压测

写一个 test.sh ：

#!/bin/bash
for i in {1..10}; do
  curl -X POST http://localhost:8000/process \
    -H "Content-Type: application/json" \
    -d '{"address":"上海市徐汇区漕溪北路'$i'号"}' &
done
wait
echo "10 concurrent requests sent"

配合 uvicorn 启动一个简易API（ src/api.py ）：

from fastapi import FastAPI
from src.main import orchestrator

app = FastAPI()

@app.post("/process")
async def process_request(input_data: dict):
    result = orchestrator.run(input_data)
    return result.model_dump()

运行： uvicorn src.api:app --reload --port 8000

此时你已经有了一个可运行、可监控、可压测的Agentic原型。接下来的所有扩展——加第二个Agent、换LLM、接数据库——都只是在这个骨架上添砖加瓦。

5. 常见问题与排查技巧实录：那些文档里不会写的坑

5.1 “Agent执行了，但Orchestrator没收到返回”——进程隔离陷阱

现象 ：在 Execution Engine 中用 multiprocessing.Process 启动Agent，日志显示Agent已执行完毕，但Orchestrator一直卡在 _execute_agent() ，最终超时。

根因 ：Python多进程间默认不共享内存。Agent子进程中的 print() 日志能输出，但 ExecutionResult 对象无法自动序列化回父进程。我们曾用 pickle 尝试传递，结果发现 requests.Response 对象不可序列化，直接报错。

解决方案 ：强制使用 multiprocessing.Queue 进行进程间通信，且只传递基础数据类型：

def _execute_in_process(self, agent_id: str, input_dict: dict, 
                       result_queue: Queue, timeout_sec: int):
    try:
        agent = self.agents[agent_id]
        result = agent.execute(input_dict)
        # 只传递dict，不传递ExecutionResult对象
        result_queue.put({
            "status": result.status,
            "output": result.output,
            "error": result.error,
            "metrics": result.metrics
        })
    except Exception as e:
        result_queue.put({
            "status": "failed",
            "error": str(e),
            "output": {}
        })

# 在_orchestrator中调用
result_queue = Queue()
proc = Process(target=self._execute_in_process, 
               args=(invocation.agent_id, invocation.input, 
                     result_queue, invocation.timeout_sec))
proc.start()
proc.join(timeout_sec)
if proc.is_alive():
    proc.terminate()
    proc.join()
    return ExecutionResult.timeout("Process killed by timeout")
    
if not result_queue.empty():
    raw_result = result_queue.get()
    return ExecutionResult(**raw_result)
else:
    return ExecutionResult.error("No result from process")

实操心得：这个坑我们踩了整整两天。后来发现，连 datetime 对象都不能直接放Queue里（需转为ISO字符串），所有Agent的 output 字段必须是JSON可序列化的 dict/list/str/int/float/None 。我们在 ExecutionResult.success() 方法里加了强制校验： json.dumps(output) ，不通过直接抛异常。

5.2 “Orchestrator决策越来越慢”——状态快照膨胀

现象 ：系统运行一周后，单次 run() 耗时从200ms涨到3.5秒， state["history"] 数组长度达1200+项。

根因 ：Orchestrator默认把每次Agent执行的完整 input 和 output 都存进 state["history"] 。而 output 里可能包含整页PDF文本、Base64图片、长SQL查询结果——这些大对象在内存中不断累积。

解决方案 ：实施三级状态裁剪策略：

1. Agent层裁剪 ：在Agent的 execute() 末尾，主动删除大字段：

   def execute(self, input_dict: dict) -> ExecutionResult:
       # ... 执行逻辑
       output = {"text": full_text, "pages": 12, "tokens": 4500}
       # 删除原始文本，只留摘要
       if len(output.get("text", "")) > 1000:
           output["text_summary"] = output["text"][:500] + "..."
           del output["text"]
       return ExecutionResult.success(output)
   

2. Orchestrator层裁剪 ：在 _update_state() 中，对 history 数组做滑动窗口：

   def _update_state(self, state, invocation, result):
       # 只保留最近5次执行记录
       state["history"] = (state["history"] + [record])[-5:]
       return state
   

3. 持久化层卸载 ：将完整 history 异步写入Redis或数据库， state 中只存 history_id ：

   # 写入Redis
   redis_client.setex(f"history:{state['session_id']}", 
                      3600, json.dumps(full_history))
   # state中只存
   state["history_ref"] = f"history:{state['session_id']}"
   

5.3 “LLM调度器输出格式错乱”——Prompt工程的硬核约束

现象 ：Orchestrator调用Qwen2-7B生成 AgentInvocation 时，有时返回 {"agent_id": "geo_locator", "input": {...}} ，有时返回 Agent: geo_locator\nInput: {...} ，导致 json.loads() 直接崩溃。

根因 ：LLM是概率模型，即使加了 response_format={"type": "json_object"} ，也无法100%保证。我们测试发现，Qwen2-7B在prompt末尾加 <|im_end|> 后，JSON格式服从率从92%提升到99.8%。

终极方案 ： Prompt + 正则双重保险 ：

def _parse_llm_output(self, raw_output: str) -> Optional[AgentInvocation]:
    # 第一步：用正则提取最可能的JSON块
    json_match = re.search(r'\{.*?\}', raw_output, re.DOTALL)
    if not json_match:
        return None
    
    # 第二步：尝试解析
    try:
        data = json.loads(json_match.group())
        return AgentInvocation(**data)
    except json.JSONDecodeError:
        # 第三步：用LLM自身修复（调用一次轻量修复模型）
        repair_prompt = f"""你是一个JSON修复器。请将以下非标准JSON修复为合法JSON：
        {raw_output}
        只输出修复后的JSON，不要任何解释。"""
        fixed_json = self.llm.invoke(repair_prompt)
        try:
            return AgentInvocation(**json.loads(fixed_json))
        except:
            return None

注意：这个修复调用必须设置极短超时（2秒）和极低温度（0.1），且失败时直接返回None，触发Orchestrator的fallback逻辑。我们宁可降级，也不信一个不稳定的JSON解析。

5.4 生产环境避坑清单（血泪总结）

问题	表象	解决方案	我们的实践
Agent内存泄漏	连续运行24小时后，进程RSS内存从300MB涨到2.1GB	在Execution Engine中强制 psutil.Process().memory_info().rss 监控，超阈值 os.kill(os.getpid(), signal.SIGTERM)	为每个Agent进程设置 MEMORY_LIMIT_MB=512 ，超限即重启
时区混乱	日志时间戳显示UTC，但业务要求东八区	所有时间相关操作统一用 datetime.now(pytz.timezone('Asia/Shanghai'))	在 Orchestrator.__init__() 中设置 os.environ['TZ'] = 'Asia/Shanghai' ，然后 time.tzset()
**