企业官网建设流程全解析

1. 项目概述：为什么需要手写一个两层嵌套循环的 Sub-Agent 架构？

“手写 Sub-Agent：两层嵌套循环实现主从 Agent 协作”——这个标题里藏着当前 LLM 应用开发中最真实、也最容易被掩盖的痛点：不是所有 Agent 框架都适合你的业务逻辑，更不是所有“开箱即用”的抽象层都能扛住生产级的复杂度和可控性要求。我过去三年在金融风控、政务知识库、工业设备诊断三个垂直领域落地过 17 个 LLM Agent 项目，90% 的失败案例，根源不在模型能力，而在于Agent 编排层的黑盒化与失控感。比如你用 LangChain 的AgentExecutor跑一个带工具调用的 RAG 流程，一旦中间某次tool_calls返回格式异常、或工具执行超时、或 LLM 在 ReAct 步骤中擅自跳过思考直接输出答案——整个链路就断了，日志里只有一行Failed to parse action input，你根本不知道是 prompt 写错了、tool schema 定义松了、还是 LLM 自己“编造”了参数。而这个项目标题所指的“手写”，本质是把 Agent 的决策流、工具调度流、状态流转全部收回到开发者手中，用最朴素的 Python 循环+条件判断+结构化状态管理，构建可调试、可拦截、可审计的主从协作骨架。

核心关键词 “Sub-Agent” 在这里不是指某个框架里的内置类，而是指一种明确的角色划分：主 Agent（Orchestrator）负责全局目标拆解、任务分派、结果聚合与终局判断；子 Agent（Worker）专注单一能力域——比如一个专做 SQL 查询的子 Agent，一个专做文档摘要的子 Agent，一个专做外部 API 校验的子 Agent。它们之间不共享内存，不隐式通信，所有交互必须通过明确定义的输入/输出 Schema 和状态上下文传递。而“两层嵌套循环”正是这种协作的物理实现：外层循环控制主 Agent 的 ReAct 迭代（Thought-Action-Observation），内层循环则封装子 Agent 的完整生命周期——从接收任务、选择工具、构造调用、等待响应、到解析结果并返回结构化 payload。这不是炫技，而是为了在每一个环节插入检查点：比如在内层循环开始前校验子 Agent 的输入合法性，在工具调用后强制验证返回 JSON 是否符合预期 Schema，在 Observation 解析失败时触发降级策略（如重试、换工具、或抛出结构化错误码）。这种设计让整个系统像一台精密的瑞士手表——每个齿轮的转速、咬合时机、受力反馈都清晰可见，而不是一个混沌的、靠概率运行的黑箱。

它解决的不是“能不能跑起来”的问题，而是“出了问题能不能三分钟定位根因”“上线后能不能按需开关某个子能力”“审计时能不能导出每一步的原始输入输出”这些真正在意的问题。适合谁？如果你正在评估是否要引入 AutoGen、LangGraph 或 LlamaIndex 的 Agent 模块，但团队里没有专职 Prompt 工程师；如果你的业务涉及多步骤合规审批，每一步都需要留痕和人工复核；如果你的工具链包含私有 API、数据库直连、甚至硬件串口指令——那么这套手写方案不是备选，而是起点。它不追求“最先进”，但追求“最可控”。接下来，我会带你从零开始，用纯 Python 实现这个架构，不依赖任何第三方 Agent 框架，只基于标准库和openai/anthropic等主流 LLM SDK，所有代码均可直接粘贴运行，所有设计决策背后都有我在产线踩过的坑作为依据。

2. 整体架构设计与思路拆解：为什么是两层循环，而不是状态机或事件驱动？

2.1 主从角色定义与职责边界必须物理隔离

在开始写代码前，先厘清一个关键认知：Sub-Agent 不是“更小的 Agent”，而是“有明确边界的专用计算单元”。很多团队误以为把一个大 Agent 拆成几个小 Agent 就是“模块化”，结果只是把耦合从代码层面转移到了 prompt 层面——主 Agent 的 prompt 里硬编码了子 Agent 的能力描述，子 Agent 的输出又依赖主 Agent 的上下文注入，最终形成环状依赖，调试时牵一发而动全身。本方案采用物理隔离原则：主 Agent 和子 Agent 之间只通过 JSON Schema 交换数据，不共享任何对象引用、不共用任何全局变量、不隐式继承任何父类。主 Agent 的输出是一个严格定义的TaskAssignment对象，包含worker_type（子 Agent 类型标识）、input_data（序列化后的输入）、timeout_sec（超时阈值）；子 Agent 的输入必须是该 Schema 的完整实例，输出必须是WorkerResult，包含status（success/error）、payload（结构化结果）、log（执行过程快照）。这种设计带来的第一个好处是可独立测试：你可以单独启动一个 SQL 子 Agent，用 Postman 发送{"worker_type":"sql_executor","input_data":{"query":"SELECT * FROM users WHERE id=123"}}，立刻验证其行为，无需启动整个主流程。

第二个好处是部署弹性。当你的 SQL 子 Agent 因为数据库连接池耗尽而卡死时，主 Agent 可以在内层循环的 timeout 判定后，立即降级到备用方案（比如改用缓存查询），而不会拖垮整个主流程。如果采用共享内存的状态机设计，一个子模块的阻塞会冻结整个状态机线程。我们选择两层循环而非单层状态机，正是因为循环天然具备“中断-恢复-重试”的语义：外层循环的每次迭代代表一次完整的 ReAct 周期，内层循环的每次迭代代表一次子任务的端到端执行，两者生命周期正交，互不干扰。

2.2 两层循环的物理实现与控制流设计

外层循环（主 Agent 循环）的核心逻辑是经典的 ReAct 模式，但做了关键增强：

# 外层循环伪代码（实际代码见第3节） while not is_final_answer(): # Step 1: 主 Agent 思考下一步行动 thought, action_name, action_input = orchestrator.think(current_context) # Step 2: 根据 action_name 分派给对应子 Agent if action_name in WORKER_REGISTRY: # 构建子任务分配对象 task = TaskAssignment( worker_type=action_name, input_data=action_input, timeout_sec=WORKER_TIMEOUTS.get(action_name, 30) ) # Step 3: 启动内层循环执行子任务 result = execute_worker_loop(task) # Step 4: 将子任务结果注入上下文，供下次思考使用 current_context = update_context_with_observation( current_context, action_name, result ) else: raise ValueError(f"Unknown action: {action_name}")

内层循环（子 Agent 循环）则聚焦于“可靠执行”：

# 内层循环伪代码 def execute_worker_loop(task: TaskAssignment) -> WorkerResult: for attempt in range(MAX_RETRIES): try: # Step 1: 实例化对应子 Agent（物理隔离！） worker = WORKER_REGISTRY[task.worker_type]() # Step 2: 验证输入合法性（防御第一关） if not worker.validate_input(task.input_data): return WorkerResult(status="error", payload={"reason": "invalid_input"}) # Step 3: 执行核心逻辑（可能包含多次工具调用） result = worker.execute(task.input_data) # Step 4: 强制 Schema 校验（防御第二关） if not worker.validate_output(result): raise ValidationError("Output does not match expected schema") return WorkerResult(status="success", payload=result, log=worker.get_execution_log()) except TimeoutError: log_warning(f"Worker {task.worker_type} timeout on attempt {attempt}") continue except ValidationError as e: log_error(f"Worker {task.worker_type} output validation failed: {e}") break # 不重试，Schema 错误是设计问题，非临时故障 except Exception as e: log_error(f"Worker {task.worker_type} execution failed: {e}") continue return WorkerResult(status="error", payload={"reason": "max_retries_exceeded"})

这个设计的关键在于：外层循环管“做什么”，内层循环管“怎么做”和“做没做好”。主 Agent 不关心子 Agent 内部如何连接数据库、如何解析 JSON、如何重试 HTTP 请求——它只认WorkerResult这个契约。而子 Agent 也不需要理解主 Agent 的全局目标，它只处理TaskAssignment中明确指定的输入。这种解耦让每个模块都可以独立演进：你可以把 SQL 子 Agent 的底层驱动从psycopg2升级到asyncpg，只要execute()方法的输入输出 Schema 不变，主流程完全无感；你也可以为主 Agent 增加新的 ReAct 策略（比如加入反思机制），而无需改动任何一个子 Agent 的代码。

2.3 为什么不用事件驱动或协程？——面向运维的务实选择

看到这里，你可能会问：为什么不用asyncio做并发？为什么不用 Kafka 做任务队列？为什么不用状态机库（如transitions）？答案很实在：在中小规模、低并发、强一致性要求的业务场景下，同步阻塞式两层循环的可维护性远高于异步或事件驱动方案。我曾在某省级政务知识库项目中尝试用asyncio.gather并发调用多个子 Agent，结果发现：当某个子 Agent 因网络抖动延迟 5 秒时，整个主流程的await会被阻塞，而其他本可并行的子任务却在空等；更糟的是，asyncio的错误堆栈极其晦涩，一次数据库连接超时会引发长达 20 行的嵌套RuntimeWarning，定位问题耗时是同步方案的 3 倍。而 Kafka 方案则引入了额外的运维成本——你需要维护 ZooKeeper、监控 Topic Lag、处理消费者组 rebalance，对于一个只有 3 个子 Agent、QPS 不到 5 的内部系统，这是典型的“杀鸡用牛刀”。

两层循环的优势在于调试友好性：你在 IDE 里打个断点，就能清晰看到外层循环的current_context如何一步步演化，内层循环的attempt计数器如何递增，result.payload的字段如何被填充。日志打印也极其直观：“[Orchestrator] Iteration 3: thinking... [SQL_Worker] Attempt 1: executing query... [SQL_Worker] Success: got 12 rows”。这种线性、可追溯的执行流，是保障线上系统稳定性的基石。当然，这不是否定异步的价值——当你需要同时调度 50 个子 Agent 处理批量文件解析时，asyncio是必选项。但本项目的目标是提供一个最小可行、最大可控的基线实现，让开发者能亲手触摸到 Agent 协作的每一根神经，而不是被框架的抽象层隔绝在外。

3. 核心细节解析与实操要点：从 Schema 设计到工具调用封装

3.1 严格 Schema 定义：用 Pydantic v2 构建不可绕过的契约

所有跨模块通信的数据结构，必须用 Pydantic v2 的BaseModel显式定义。这不是为了“类型安全”的虚名，而是为了在数据流入流出的第一时间捕获错误。我们定义三个核心 Schema：

from pydantic import BaseModel, Field, validator from typing import Dict, Any, Optional, List import re class TaskAssignment(BaseModel): """主 Agent 分派给子 Agent 的任务契约""" worker_type: str = Field(..., description="子 Agent 类型标识，如 'sql_executor' 或 'web_searcher'") input_data: Dict[str, Any] = Field(..., description="序列化后的输入数据，必须符合子 Agent 的 input_schema") timeout_sec: int = Field(30, ge=1, le=300, description="子任务超时时间，单位秒") @validator('worker_type') def validate_worker_type(cls, v): if not re.match(r'^[a-z][a-z0-9_]*$', v): raise ValueError('worker_type must be snake_case and start with letter') return v class WorkerResult(BaseModel): """子 Agent 执行结果的标准化输出""" status: str = Field(..., pattern=r'^(success|error)$', description="执行状态") payload: Dict[str, Any] = Field(..., description="结构化结果数据，成功时为业务数据，失败时为错误详情") log: Optional[Dict[str, Any]] = Field(None, description="执行过程快照，用于调试") class OrchestratorContext(BaseModel): """主 Agent 的全局上下文状态""" conversation_history: List[Dict[str, str]] = Field(default_factory=list) current_goal: str = Field(..., description="当前待解决的顶层目标") intermediate_results: Dict[str, Any] = Field(default_factory=dict, description="已获取的中间结果，键为 action_name") iteration_count: int = Field(0, ge=0, description="ReAct 迭代次数") def add_message(self, role: str, content: str): self.conversation_history.append({"role": role, "content": content}) def get_last_user_message(self) -> str: for msg in reversed(self.conversation_history): if msg["role"] == "user": return msg["content"] return ""

关键细节说明：

• TaskAssignment.worker_type的@validator强制要求蛇形命名且首字符为字母，这避免了因拼写错误（如SQL_Executorvssql_executor）导致的注册表查找失败。
• WorkerResult.status使用pattern而非Literal，是因为Literal在 Pydantic v2 中序列化时会生成冗余的enum字段，而pattern更轻量且语义清晰。
• OrchestratorContext提供了add_message和get_last_user_message方法，将状态操作封装为方法，避免外部代码直接修改conversation_history导致的并发问题（虽然本方案是同步单线程，但为未来扩展预留接口）。

提示：所有子 Agent 的execute()方法签名必须为def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]，返回值将被自动包装进WorkerResult.payload。这样设计是为了让子 Agent 开发者只需关注业务逻辑，Schema 封装由框架层统一处理。

3.2 子 Agent 工具调用的封装范式：从 raw API 到可插拔工具链

子 Agent 的核心价值在于封装工具调用的复杂性。以最常见的 SQL 查询子 Agent 为例，它的execute()方法内部可能涉及：连接池管理、SQL 注入防护、查询超时控制、结果集截断、JSON 序列化。但对外暴露的，只是一个干净的input_data字典：

class SQLExecutor: def __init__(self): # 连接池初始化（生产环境应使用 connection pool） self.engine = create_engine( "postgresql://user:pass@localhost/db", pool_pre_ping=True, # 每次使用前检测连接有效性 pool_recycle=3600 # 连接复用 1 小时后强制回收 ) def validate_input(self, input_data: Dict[str, Any]) -> bool: # 关键校验：防止恶意 SQL if not isinstance(input_data, dict) or "query" not in input_data: return False query = input_data["query"].strip() # 禁止 DDL 和危险操作 forbidden_patterns = [r'\b(DROP|CREATE|ALTER|TRUNCATE)\b', r';\s*$', r'--'] return not any(re.search(pattern, query, re.IGNORECASE) for pattern in forbidden_patterns) def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]: query = input_data["query"].strip() limit = input_data.get("limit", 100) # 默认限制 100 行 with self.engine.connect() as conn: # 使用 text() 包装，避免 ORM 开销 stmt = text(f"{query} LIMIT :limit") result = conn.execute(stmt, {"limit": limit}) # 结果集转换为字典列表 columns = result.keys() rows = [dict(zip(columns, row)) for row in result.fetchall()] return { "query": query, "row_count": len(rows), "data": rows[:1000], # 强制截断，防 OOM "execution_time_ms": result._metadata.execution_time_ms if hasattr(result, '_metadata') else 0 } def validate_output(self, output: Dict[str, Any]) -> bool: # 必须包含期望字段 required_keys = {"query", "row_count", "data"} return required_keys.issubset(output.keys()) and isinstance(output["data"], list)

这个实现体现了三个关键原则：

1. 输入防御：validate_input在执行前就过滤掉高危 SQL，比在数据库层拦截更早、更安全；
2. 资源管控：LIMIT参数由主 Agent 注入，子 Agent 无权突破，避免全表扫描；
3. 输出契约：validate_output确保返回结构始终一致，主 Agent 可以放心地result.payload["data"][0]["name"]访问字段。

其他子 Agent（如 Web Searcher、Document Summarizer）都遵循同一范式：validate_input→execute→validate_output。这种一致性让主 Agent 的调度逻辑变得极其简单——它不需要为每个子 Agent 写不同的解析代码，只需要统一处理WorkerResult.payload。

3.3 主 Agent 的 ReAct Prompt 工程：用结构化输出规避 LLM 的“自由发挥”

主 Agent 的 prompt 是整个系统的“大脑”，但它不能依赖 LLM 的“常识”。我们采用XML 标签 + JSON Schema 强约束的方式，让 LLM 的输出可预测：

ORCHESTRATOR_SYSTEM_PROMPT = """ 你是一个严谨的 AI 助理，负责根据用户问题规划执行步骤。请严格按以下规则响应： 1. 你的输出必须是纯 JSON，不含任何 Markdown、注释或额外文本。 2. 输出必须包含三个字段：{"thought": "string", "action": "string", "action_input": "object"} 3. "action" 必须是以下之一：["sql_executor", "web_searcher", "document_summarizer", "final_answer"] 4. "action_input" 的结构必须严格匹配对应 action 的 input_schema： - sql_executor: {"query": "valid SQL SELECT statement", "limit": "integer"} - web_searcher: {"query": "search keywords", "site": "optional domain filter"} - document_summarizer: {"text": "plain text to summarize", "max_length": "integer"} - final_answer: {"answer": "final response to user"} 现在开始。用户问题是：{user_question} """ def generate_orchestrator_prompt(context: OrchestratorContext) -> str: # 将上下文历史压缩为字符串，保留最后 3 轮对话 history_str = "\n".join([ f"{msg['role']}: {msg['content'][:200]}..." for msg in context.conversation_history[-3:] ]) return ORCHESTRATOR_SYSTEM_PROMPT.format(user_question=history_str)

关键技巧：

• 禁止自由文本：明确要求“纯 JSON，不含任何 Markdown”，避免 LLM 在 JSON 外围添加解释性文字；
• 枚举 action 名称：强制 LLM 只能从预设列表中选择，杜绝拼写错误；
• Schema 内联提示：在 prompt 中直接写出每个 action 的action_input结构，比单独维护一个 schema 文件更可靠（因为 LLM 看不到外部文件）；
• 上下文截断：context.conversation_history[-3:]限制历史长度，防止 prompt 过长导致 token 超限。

实测下来，这种 prompt 在 GPT-4-turbo 上的 JSON 解析成功率从 82% 提升到 99.3%，失败案例几乎全是action字段拼写错误（如"sql_executer"），而这类错误可以通过WORKER_REGISTRY的 key 检查在代码层兜底。

4. 实操过程与核心环节实现：从零开始搭建可运行的两层循环

4.1 环境准备与依赖安装（极简主义）

本项目仅依赖三个包，全部来自 PyPI 官方源，无任何魔改或 beta 版本：

pip install pydantic==2.7.1 openai==1.35.1 python-dotenv==1.0.1

• pydantic==2.7.1：稳定版，v2 的成熟分支，避免 v2.8+ 的 breaking changes；
• openai==1.35.1：兼容 OpenAI 官方 API 的最新稳定版，支持chat.completions流式响应；
• python-dotenv==1.0.1：安全读取.env文件，避免 API Key 硬编码。

注意：不要安装langchain、llamaindex或autogen。本方案的哲学是“用最少的依赖，做最多的事”。所有 Agent 编排逻辑均由我们自己编写，不引入任何框架的抽象层。

4.2 完整可运行代码：主循环、子 Agent 注册、LLM 调用封装

以下是经过生产环境验证的完整代码（已移除敏感信息，可直接保存为sub_agent_orchestrator.py）：

import os import json import time import logging from typing import Dict, Any, List, Optional from pydantic import BaseModel, Field, validator from openai import OpenAI # 配置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # 加载环境变量 from dotenv import load_dotenv load_dotenv() # 初始化 OpenAI 客户端 client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) # === Schema 定义（同第3节，此处省略重复代码）=== # === 子 Agent 注册表 === WORKER_REGISTRY = {} WORKER_TIMEOUTS = { "sql_executor": 45, "web_searcher": 30, "document_summarizer": 60, } def register_worker(worker_type: str, worker_class): """注册子 Agent 类""" if worker_type in WORKER_REGISTRY: raise ValueError(f"Worker type '{worker_type}' already registered") WORKER_REGISTRY[worker_type] = worker_class logger.info(f"Registered worker: {worker_type}") # === 子 Agent 实现 === class SQLExecutor: def __init__(self): # 生产环境应替换为连接池 from sqlalchemy import create_engine, text self.engine = create_engine("sqlite:///example.db") # 示例用 SQLite def validate_input(self, input_data: Dict[str, Any]) -> bool: if not isinstance(input_data, dict) or "query" not in input_data: return False query = input_data["query"].strip().upper() return not any(kw in query for kw in ["DROP", "CREATE", "ALTER", "TRUNCATE", "DELETE"]) def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]: query = input_data["query"] limit = input_data.get("limit", 10) with self.engine.connect() as conn: result = conn.execute(text(f"{query} LIMIT :limit"), {"limit": limit}) columns = result.keys() rows = [dict(zip(columns, row)) for row in result.fetchall()] return {"query": query, "row_count": len(rows), "data": rows} def validate_output(self, output: Dict[str, Any]) -> bool: return all(k in output for k in ["query", "row_count", "data"]) # 注册子 Agent register_worker("sql_executor", SQLExecutor) # === 内层循环实现 === MAX_RETRIES = 3 def execute_worker_loop(task: TaskAssignment) -> WorkerResult: """执行子 Agent 的内层循环""" worker_class = WORKER_REGISTRY.get(task.worker_type) if not worker_class: return WorkerResult( status="error", payload={"reason": f"Unknown worker_type: {task.worker_type}"} ) for attempt in range(MAX_RETRIES): try: worker = worker_class() if not worker.validate_input(task.input_data): return WorkerResult( status="error", payload={"reason": "Invalid input data"} ) # 执行核心逻辑 start_time = time.time() result = worker.execute(task.input_data) elapsed = int((time.time() - start_time) * 1000) if not worker.validate_output(result): raise ValueError("Output validation failed") return WorkerResult( status="success", payload=result, log={"execution_time_ms": elapsed, "attempt": attempt + 1} ) except Exception as e: logger.warning(f"Worker {task.worker_type} failed on attempt {attempt + 1}: {e}") if attempt == MAX_RETRIES - 1: return WorkerResult( status="error", payload={"reason": str(e), "attempt": attempt + 1} ) time.sleep(0.5 * (2 ** attempt)) # 指数退避 return WorkerResult(status="error", payload={"reason": "Unexpected error"}) # === 主 Agent 循环实现 === def is_final_answer(payload: Dict[str, Any]) -> bool: """判断是否为最终答案""" return payload.get("action") == "final_answer" def update_context_with_observation( context: OrchestratorContext, action_name: str, worker_result: WorkerResult ) -> OrchestratorContext: """将子任务结果注入上下文""" if worker_result.status == "success": context.intermediate_results[action_name] = worker_result.payload context.add_message("assistant", f"Executed {action_name}: success") else: context.add_message("assistant", f"Executed {action_name}: failed - {worker_result.payload.get('reason', 'unknown')}") return context def orchestrator_think(context: OrchestratorContext) -> Dict[str, Any]: """主 Agent 的思考环节：调用 LLM 生成下一步动作""" prompt = generate_orchestrator_prompt(context) try: response = client.chat.completions.create( model="gpt-4-turbo", messages=[{"role": "system", "content": prompt}], temperature=0.1, # 降低随机性 max_tokens=512, response_format={"type": "json_object"} # 强制 JSON 输出 ) raw_content = response.choices[0].message.content.strip() return json.loads(raw_content) except Exception as e: logger.error(f"LLM call failed: {e}") raise def run_orchestration(user_question: str, max_iterations: int = 10) -> str: """启动主从协作的两层循环""" context = OrchestratorContext( conversation_history=[{"role": "user", "content": user_question}], current_goal=user_question, intermediate_results={}, iteration_count=0 ) for iteration in range(max_iterations): context.iteration_count = iteration + 1 logger.info(f"--- Iteration {iteration + 1} ---") # Step 1: 主 Agent 思考 try: llm_output = orchestrator_think(context) except Exception as e: logger.error(f"Orchestrator think failed: {e}") return f"Orchestration failed at iteration {iteration + 1}: {e}" # Step 2: 解析 LLM 输出 if not isinstance(llm_output, dict) or "action" not in llm_output: logger.error(f"Invalid LLM output: {llm_output}") return "LLM returned invalid format" action_name = llm_output["action"] action_input = llm_output.get("action_input", {}) # Step 3: 判断是否为最终答案 if is_final_answer(llm_output): return llm_output["action_input"]["answer"] # Step 4: 分派子任务 if action_name not in WORKER_REGISTRY: logger.error(f"Unknown action: {action_name}") return f"Unknown action: {action_name}" task = TaskAssignment( worker_type=action_name, input_data=action_input, timeout_sec=WORKER_TIMEOUTS.get(action_name, 30) ) # Step 5: 执行内层循环 worker_result = execute_worker_loop(task) # Step 6: 更新上下文 context = update_context_with_observation(context, action_name, worker_result) # Step 7: 检查是否失败过多 if worker_result.status == "error": logger.warning(f"Worker {action_name} failed: {worker_result.payload}") if iteration >= max_iterations - 1: return f"Failed after {max_iterations} iterations: {worker_result.payload.get('reason', 'unknown')}" return "Max iterations reached without final answer" # === 使用示例 === if __name__ == "__main__": # 创建示例数据库（SQLite） import sqlite3 conn = sqlite3.connect("example.db") conn.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER, name TEXT, email TEXT)") conn.execute("INSERT OR REPLACE INTO users VALUES (1, 'Alice', 'alice@example.com'), (2, 'Bob', 'bob@example.com')") conn.commit() conn.close() # 运行示例 result = run_orchestration("查询用户表中所有人的姓名和邮箱") print("Final Result:", result)

这段代码的关键实操细节：

• response_format={"type": "json_object"}：OpenAI API 的原生 JSON 模式，比在 prompt 里写“请输出 JSON”更可靠，失败率降低 40%；
• temperature=0.1：极低温度值，确保 LLM 输出高度确定，避免在 ReAct 步骤中“灵光一现”；
• time.sleep(0.5 * (2 ** attempt))：指数退避重试，第一次失败后等 0.5 秒，第二次等 1 秒，第三次等 2 秒，避免对下游服务造成雪崩；
• SQLite 示例数据库：用sqlite3内置模块快速搭建测试环境，无需额外数据库服务。

4.3 运行效果与典型输出分析

执行上述代码，你会看到类似这样的日志输出：

2024-06-15 10:23:42,123 - INFO - Registered worker: sql_executor 2024-06-15 10:23:42,124 - INFO - --- Iteration 1 --- 2024-06-15 10:23:42,125 - INFO - Calling LLM... 2024-06-15 10:23:45,678 - INFO - LLM returned: {'thought': '需要查询 users 表获取姓名和邮箱', 'action': 'sql_executor', 'action_input': {'query': 'SELECT name, email FROM users', 'limit': 100}} 2024-06-15 10:23:45,679 - INFO - Executing sql_executor... 2024-06-15 10:23:45,789 - INFO - Worker sql_executor succeeded in 110ms 2024-06-15 10:23:45,790 - INFO - Executed sql_executor: success 2024-06-15 10:23:45,791 - INFO - --- Iteration 2 --- 2024-06-15 10:23:45,792 - INFO - Calling LLM... 2024-06-15 10:23:48,345 - INFO - LLM returned: {'thought': '已获取用户数据，可以生成最终回答', 'action': 'final_answer', 'action_input': {'answer': '用户信息如下：Alice (alice@example.com), Bob (bob@example.com)'}} Final Result: 用户信息如下：Alice (alice@example.com), Bob (bob@example.com)

这个输出清晰展示了两层循环的协作：Iteration 1 中，主 Agent 决定调用sql_executor，内层循环执行查询并返回结果；Iteration 2 中，主 Agent 基于上一轮的intermediate_results，决定输出最终答案。整个过程没有魔法，每一步都可追踪、可打断、可修改。

5. 常见问题与排查技巧实录：那些只有亲手写过才懂的坑

5.1 LLM 输出 JSON 解析失败：90% 的问题出在 prompt，而非模型

现象：json.loads(raw_content)抛出JSONDecodeError，错误信息为Expecting property name enclosed in double quotes。

原因分析：LLM 在输出 JSON 时，有时会用单引号包裹 key（如{'thought': 'xxx'}），或在 JSON 外围添加解释性文字（如Here is the JSON: {"thought": "xxx"}）。这不是模型能力问题，而是 prompt 约束力不足。

解决方案：

• 强化 prompt 的 JSON 约束：在 system prompt 开头增加一行Output ONLY valid JSON. No explanations, no markdown, no extra characters.；
• 增加 JSON 清洗逻辑：在orchestrator_think函数中，添加容错解析：

def safe_json_loads(s: str) -> Dict[str, Any]: # 移除首尾空白和常见前缀 s = s.strip() if s.startswith("```json"): s = s[7:] if s.endswith("```"): s = s[:-3] # 尝试修复单引号 s = s.replace("'", '"') # 移