1. 项目缘起：当“多智能体”成为标配，我们为何还要造轮子？

去年下半年开始，整个AI开发圈的氛围变得有点微妙。如果你参加过任何一场技术分享会，或者浏览过GitHub的趋势榜，你会发现“多智能体”（Multi-Agent）这个词出现的频率，几乎和当年的“微服务”一样高。一夜之间，仿佛所有复杂的AI应用问题，都可以通过“让几个AI智能体互相聊聊天”来解决。从自动化的代码审查、复杂的客服流程编排，到动态的市场数据分析，多智能体架构似乎成了那把“万能钥匙”。

我们也跟上了这股潮流。团队接到了一个为大型金融机构构建内部智能分析助手的项目，核心需求是让AI能够自主完成从数据提取、清洗、分析到生成报告和风险预警的全链条任务。这显然不是单个大语言模型（LLM）调用能搞定的，一个“智能体”干不了，那就上一群——这很符合多智能体的设计哲学。于是，和所有技术团队一样，我们的第一反应是：找现成的框架。

在接下来的三个月里，我们进行了一场密集的“框架选型马拉松”。从早期以对话编排见长的 AutoGen ，到强调简易工作流设计的 CrewAI ，再到功能全面、社区活跃的 LangGraph ，我们几乎把市面上所有叫得上名号的开源与商业框架都试了个遍。我们按照官方教程搭建原型，模拟真实业务场景进行压力测试，甚至深入源码去理解它们的调度逻辑和通信机制。

这个过程很有收获，但也让我们陷入了一种普遍的“框架困境”。我们发现，这些框架在展示其设计理念和基础能力时都很棒，但一旦将它们塞进我们那个充满特定业务规则、复杂状态管理和苛刻性能要求的真实项目里时，各种“水土不服”就出现了。有的框架为了通用性，抽象层级太高，导致我们在实现一个具体的业务状态校验时，需要写大量的胶水代码；有的框架在智能体间通信上采用了固定的模式，无法灵活支持我们需要的“同步-异步混合”的消息传递；还有的框架在任务失败重试、资源隔离和观测性（Observability）方面非常薄弱，而这对于企业级应用来说是不可接受的。

最终，我们做出了一个看似“反潮流”的决定：不再试图勉强适配任何一个现有框架，而是基于我们踩过的所有坑和提炼出的核心需求，从头构建一个属于我们自己的多智能体运行时（Runtime）。这不是为了炫技，而是一次被真实需求“逼出来”的深度实践。这篇文章，就是这次“造轮子”之旅的完整复盘，我会详细拆解我们为何放弃现成方案，以及我们如何设计并实现一个更贴合复杂业务场景的轻量级、高性能多智能体运行时核心。

2. 现有框架深度体验：理想与现实的落差

在决定自研之前，我们对主流框架进行了近乎“苛刻”的评估。我们的评估维度不仅包括易用性和功能完整性，更聚焦于 生产就绪性 ：这个框架能否在高压、复杂的真实业务环境中稳定、高效、可控地运行？以下是我们对几个代表性框架的深度体验与核心痛点分析。

2.1 抽象泄漏与“胶水代码”困境

以 CrewAI 为例，它的“任务-智能体-工具”三层抽象非常清晰，对于快速构建一个概念验证（PoC）来说体验极佳。你可以在半小时内就搭出一个由“研究员”、“分析师”、“撰稿人”智能体协作的新闻简报生成流水线。然而，当我们需要实现这样一个场景时，问题来了： “分析师”智能体在生成图表后，需要将图表文件路径和关键数据结论，以一种特定的结构化格式（而不仅仅是自然语言描述）传递给“质检员”智能体进行合规性校验。

在CrewAI的标准流程中，智能体间的通信主要依靠在任务描述（Task Description）中“期待”上一个智能体的输出。这本质上是基于自然语言的、隐式的约定。为了实现结构化数据传递，我们不得不做两件事：

1. 修改“分析师”智能体的提示词（Prompt），强制其输出一个包含 chart_path 和 key_metrics 的JSON字符串。
2. 在“质检员”智能体的任务中，编写额外的解析逻辑，从上游输出的文本中提取出这个JSON块。

这仅仅是两个智能体之间的一次传递。当流程涉及多个步骤、多种数据类型（文本、数据、文件引用、状态码）时，这种“胶水代码”会指数级增长，并且散落在各个智能体的提示词或工具函数里，变得难以维护和调试。 框架提供的优雅抽象“泄漏”了，我们不得不频繁地处理底层的数据表示和传递问题。 这让我们意识到，一个生产级的运行时，必须将“结构化消息”作为一等公民来支持。

2.2 调度策略的“非黑即白”之痛

另一个深刻的教训来自对调度模式的探索。 AutoGen 提供了强大的群聊（GroupChat）功能，其基于LLM的“调度器”（Scheduler）可以动态决定下一个该谁说话，这在开放式讨论中非常有趣。但我们的业务流是高度结构化的，更像一个工作流（Workflow），其中80%的步骤是预先定义好的顺序执行或条件分支。

当我们尝试用AutoGen实现一个带条件判断的流程时，比如“如果风险评分>阈值，则执行A分析，否则执行B分析”，我们面临两个选择：

1. 滥用调度器 ：将所有逻辑写在提示词里，让LLM调度器去“理解”并决定下一步。这导致了不可预测的延迟（每次调度都是一次LLM调用），且流程逻辑变得晦涩难懂，深藏在提示词的英文描述中。
2. 引入外部状态机 ：在AutoGen系统外，用Python代码维护一个状态机，然后通过监听消息和调用 group_chat.select_speaker 来强行控制流程。这相当于用另一个系统来驱动框架，架构变得复杂和冗余。

LangGraph 的图（Graph）模型在这方面是巨大的进步，它明确地用“边”来定义控制流。但是，LangGraph的图是静态编译的，虽然在定义时支持条件边（Conditional Edge），但其条件判断依赖于一个中心化的、LLM驱动的“路由函数”（Routing Function）。对于简单的“if-else”，这没问题。但对于我们场景中复杂的、基于精确业务规则（例如，“报表类型为‘季度’且数据源版本大于2.0”）的多路分支，让LLM去做判断不仅成本高，而且可能因为提示词描述的不精确而产生错误路由。

注意 ：这里暴露的核心问题是“决策逻辑的归属”。业务规则是确定性的、精确的，应该由代码来处理；而涉及语义理解、创意生成的环节，才应该交给LLM。优秀的运行时应该能清晰、方便地支持这两种决策模式的混合。

2.3 可观测性与运维的“荒漠”

这是压垮我们的最后一根稻草。当我们把用这些框架构建的原型部署到测试环境，尝试模拟连续处理100个分析任务时，运维团队提出了灵魂三问：

1. “当前这100个任务，各自跑到哪一步了？卡在哪个智能体了？” （实时状态追踪）
2. “那个失败的任务，具体是哪个工具调用出错了？错误信息和堆栈是什么？” （精细化错误定位）
3. “智能体A和智能体B之间的那次对话，具体的输入输出是什么？耗时多少？” （全链路审计与性能分析）

我们尴尬地发现，大多数框架的“日志”只是将智能体的对话打印到控制台。没有结构化的日志，没有关联的Trace ID，没有指标（Metrics）暴露。为了实现基本的可观测性，我们需要在每个工具函数、每个消息回调里手动埋点，将数据发送到我们的监控系统（如Prometheus+Grafana，或OpenTelemetry）。这又是一项巨大的、容易出错的工作。

一个面向生产的设计，必须将可观测性内化到运行时内核中，而不是事后补救。运维团队需要的是一个仪表盘，能清晰地展示：任务吞吐量、各环节耗时分布、错误率、以及每个智能体的“工作量”。这些，在当时的现有框架中，都需要我们自己去“搭”。

3. 自研运行时的核心设计哲学

基于上述痛点，我们明确了自研运行时（代号为 Orchestra ）的四大设计原则。这不仅仅是功能列表，更是贯穿每个设计决策的指导思想。

3.1 原则一：消息即接口，强类型优先

我们彻底摒弃了“自然语言作为主要通信媒介”的模糊做法。在Orchestra中，智能体（Agent）之间、智能体与工具（Tool）之间传递的，都是 强类型的消息对象（Message Object） 。

# 示例：一个强类型的分析结果消息
from pydantic import BaseModel, Field
from typing import List, Optional

class ChartInfo(BaseModel):
    path: str = Field(description="图表文件存储路径")
    chart_type: str = Field(description="图表类型，如‘line‘, ‘bar‘")
    title: str

class AnalysisResultMessage(BaseModel):
    """分析师智能体完成工作后发出的结构化消息"""
    task_id: str
    conclusion: str = Field(description="文本分析结论")
    charts: List[ChartInfo] = Field(default_factory=list)
    key_metrics: dict[str, float]
    confidence_score: float = Field(ge=0.0, le=1.0)
    next_step_recommendation: Optional[str] = None

这样做的好处是颠覆性的：

1. 清晰契约 ： AnalysisResultMessage 就是一个清晰的API接口。下游的“质检员”智能体在收到这个消息时，无需解析文本，可以直接通过 message.charts[0].path 访问图表路径。工具和智能体的输入输出被严格定义。
2. 自动验证 ：利用Pydantic，在消息创建时即可进行数据验证。如果 confidence_score 传了一个1.5的值，系统会立刻抛出验证错误，而不是让一个错误的数据在后续流程中引发更诡异的故障。
3. 代码提示与安全 ：在现代IDE中，你可以获得完整的代码自动补全和类型提示，开发体验和安全性大幅提升。序列化和反序列化（如存入数据库或消息队列）也变得异常简单可靠。

3.2 原则二：混合调度引擎：工作流打底，LLM点睛

我们设计了一个双层的调度系统，以解决“确定性流程”与“非确定性决策”的混合需求。

底层是显式的工作流引擎（Workflow Engine） 。我们采用并精简了 有向无环图（DAG） 的概念。开发者可以使用装饰器或YAML文件，明确定义智能体任务的执行顺序和依赖关系。

@orchestra.workflow(name="financial_analysis")
def define_analysis_flow():
    # 定义节点（即智能体任务）
    data_task = extract_data_agent.task()
    analysis_task = analyze_data_agent.task(depends_on=[data_task])
    # 条件分支：基于分析结果中的确定性字段进行路由
    high_risk_check = check_high_risk_agent.task(depends_on=[analysis_task])
    report_task = generate_report_agent.task(depends_on=[analysis_task])

    # 定义边（路由规则）
    orchestra.set_conditional_edge(
        source=analysis_task,
        condition=lambda ctx: ctx.get_result(analysis_task).risk_score > 0.8,
        true_target=high_risk_check,  # 高风险，走额外检查分支
        false_target=report_task       # 正常风险，直接生成报告
    )
    orchestra.set_edge(high_risk_check, report_task) # 检查完后仍需生成报告

在这个例子中， risk_score > 0.8 是一个确定性的业务规则判断，由Python函数执行，高效且准确。工作流引擎负责严格按照DAG推进，管理任务状态、处理重试和超时。

上层是灵活的LLM调度器（LLM Scheduler） ，但它不再承担核心的路由责任，而是作为工作流中的一种 特殊节点 存在。它被用于那些真正需要“智能”判断的场景。例如，在“报告生成”节点之后，我们可能增加一个“报告润色策略选择”节点：

@orchestra.llm_router(choices=[polish_for_executive, polish_for_analyst, no_polish])
def select_polish_strategy(report_content: str, audience_hint: str) -> str:
    prompt = f"""
    根据以下报告内容和受众提示，选择最合适的润色策略。
    报告摘要：{report_content[:500]}...
    受众：{audience_hint}
    可选策略：
    - ‘polish_for_executive‘: 面向高管，需要高度概括和强调商业洞察。
    - ‘polish_for_analyst‘: 面向分析师，需要保留技术细节和数据严谨性。
    - ‘no_polish‘: 无需润色。
    请只返回策略名称。
    """
    # 这里会调用LLM，但其输出被约束在三个明确选项内
    return llm_invoke(prompt)

这个LLM路由器的输出，会作为一个判断结果，被工作流引擎捕获，并用来决定下一步走向哪个润色智能体。这样，LLM被用在了它擅长（语义理解）且安全（输出被约束）的地方。

3.3 原则三：可观测性内建，而非外挂

我们从第一天就将可观测性作为运行时的核心模块来设计，其架构分为三层：

观测层	实现机制	输出示例/用途
链路追踪 (Tracing)	为每个外部请求（如API调用）生成唯一Trace ID，并在运行时内自动传播该ID到每一个智能体调用、工具调用和消息传递。	在Jaeger或Zipkin中可视化整个多智能体任务的完整调用树，精确看到耗时和调用关系。
指标度量 (Metrics)	在运行时内核的关键位置埋点，自动收集计数器、计时器和计量器。	Prometheus指标如： agent_invocation_total{agent="analyzer", status="success"} ， task_duration_seconds_bucket{workflow="financial_analysis"} ，用于监控吞吐量和性能。
结构化日志 (Logging)	所有日志事件都是结构化的JSON，并自动附加上下文（Trace ID, Agent Name, Task ID）。	日志 {“level”: “ERROR”, “trace_id”: “abc123”, “agent”: “validator”, “error”: “Data validation failed: ...”, “input_data”: {...}} 可直接被ELK或Loki索引和关联查询。

实现上，我们基于 OpenTelemetry 标准来构建这一能力。这意味着任何兼容OpenTelemetry的后端系统（如Jaeger, Prometheus, Loki）都可以无缝接入，运维团队无需为我们的运行时开发特定的监控插件。

3.4 原则四：轻量级内核与明确的生命周期

我们拒绝构建一个无所不包的“全家桶”框架。Orchestra运行时的核心职责非常明确：

1. 定义 与 加载 智能体、工具、工作流。
2. 执行 工作流，管理任务调度与状态。
3. 传递 强类型消息。
4. 收集 并 暴露 可观测性数据。

至于LLM的调用、向量数据库的连接、外部API的封装等，都被视为“外部资源”或“插件”。运行时通过清晰的接口与它们交互。例如，一个智能体需要调用LLM，它只需声明自己需要 LLMClient ，运行时会在初始化时通过依赖注入（DI）的方式提供。这带来了巨大的灵活性：你可以轻松替换不同的LLM提供商（OpenAI, Anthropic, 本地模型），而不需要修改智能体内部的代码。

智能体和工具都有明确的生命周期钩子（如 on_startup , on_shutdown , before_invoke , after_invoke ），方便开发者进行资源初始化和清理。

4. Orchestra运行时的关键实现剖析

有了清晰的设计哲学，我们开始着手实现。这里分享几个最关键组件的实现思路和遇到的挑战。

4.1 智能体（Agent）抽象：从“聊天者”到“工作者”

我们摒弃了将智能体视为“会聊天的黑盒”的模型，而是将其定义为一个 有状态的、可配置的工作单元 。一个典型的Orchestra智能体类如下：

from orchestra import Agent, BaseModel, Field
from typing import Any

class DataAnalyzerAgent(Agent):
    """数据分析智能体"""
    # 配置项：声明依赖的资源
    llm_client: LLMClient  # 运行时注入
    statsd_client: StatsDClient  # 运行时注入

    # 智能体的“技能”（工具），本地方法或外部工具
    @agent_tool
    def calculate_metrics(self, raw_data: list) -> dict:
        """计算核心指标，纯Python函数"""
        # ... 计算逻辑 ...
        self.statsd_client.increment(‘metrics_calculated‘) # 内建可观测性
        return {"avg": avg_value, "max": max_value}

    @agent_tool
    async def interpret_trend(self, metrics: dict) -> str:
        """调用LLM解读趋势，异步方法"""
        prompt = f"基于指标{metrics}，给出趋势分析..."
        response = await self.llm_client.chat(prompt)
        return response.content

    # 智能体的主入口方法
    async def invoke(self, input_msg: DataExtractionMessage) -> AnalysisResultMessage:
        """主业务逻辑：被工作流引擎调用的入口"""
        # 1. 使用工具处理数据
        metrics = self.calculate_metrics(input_msg.raw_data)
        trend = await self.interpret_trend(metrics)

        # 2. 构建强类型输出消息
        result = AnalysisResultMessage(
            task_id=input_msg.task_id,
            conclusion=trend,
            key_metrics=metrics,
            confidence_score=0.9
        )
        # 3. 自动记录调用日志（包含input_msg和result的摘要）
        self.log_invocation(input_msg, result)
        return result

关键设计点：

• 依赖注入 ： llm_client 和 statsd_client 在智能体初始化时由运行时容器注入，实现了控制反转，便于测试和替换。
• 同步/异步混合 ：工具方法可以是同步的（ calculate_metrics ）或异步的（ interpret_trend ），运行时统一处理，让开发者根据实际IO情况选择，最大化效率。
• 内建观测 ：在工具或 invoke 方法中，可以方便地使用注入的客户端打点或记录日志，这些数据会自动关联到当前任务链路。

4.2 工作流引擎：状态管理的艺术

工作流引擎的核心是管理每个任务实例的状态。我们为每个工作流运行（Workflow Run）创建一个 WorkflowContext 对象，它相当于一个共享的、类型安全的黑板（Blackboard）。

class WorkflowContext:
    """工作流执行上下文"""
    def __init__(self, workflow_run_id: str):
        self.run_id = workflow_run_id
        self._task_results: Dict[str, Any] = {} # 存储各任务输出
        self._global_state: Dict[str, Any] = {} # 全局共享状态
        self._execution_log: List[ExecutionEvent] = [] # 执行日志

    def set_result(self, task_id: str, result: BaseModel):
        """存储任务结果（强类型）"""
        self._task_results[task_id] = result

    def get_result(self, task_id: str, expected_type: Type[T]) -> T:
        """获取任务结果，并进行类型检查"""
        result = self._task_results.get(task_id)
        if not isinstance(result, expected_type):
            raise TypeError(f"Task {task_id} result type mismatch.")
        return result

工作流引擎的执行器（Executor）负责：

1. 解析DAG ：根据定义，计算任务依赖关系和执行路径。
2. 调度任务 ：将可运行的任务（所有依赖已完成）提交到 异步任务队列 （我们选用 asyncio 队列，并预留了集成Celery或Ray的接口以支持分布式）。
3. 管理状态 ：当一个任务完成时，将其输出的强类型消息存入 WorkflowContext ，并触发依赖它的任务进入就绪状态。
4. 处理控制流 ：执行条件判断函数（如之前提到的 risk_score > 0.8 ），根据结果决定下一步执行哪个节点。
5. 持久化 ：定期将 WorkflowContext 快照保存到数据库（如Redis或PostgreSQL），实现工作流的暂停、恢复和容错。这是实现 长时运行工作流 的关键。

4.3 消息总线：不只是管道，更是契约执行者

消息总线（Message Bus）是智能体间通信的枢纽。它的职责远超简单的发布/订阅。

1. 路由与分发 ：根据消息类型（如 AnalysisResultMessage ）和预定的订阅关系，将消息精准投递给一个或多个消费者智能体。
2. 序列化与验证 ：在消息发布前，利用Pydantic模型进行序列化和验证；在投递前，对数据进行反序列化和再次验证，确保传输过程中的数据一致性。
3. 持久化与重放 ：所有消息都可以选择性地持久化到日志存储（如Kafka或数据库）。这对于 调试 和 审计 至关重要。当用户报告一个异常结果时，我们可以通过Task ID找到完整的消息流，精确复现问题场景。
4. 死信队列（DLQ）处理 ：如果某个智能体处理消息连续失败，消息会被移入死信队列，防止错误消息阻塞整个系统，并触发告警通知开发者介入。

# 简化的消息总线发布示例
async def publish_message(message: BaseModel):
    # 1. 验证
    message.model_validate()
    # 2. 序列化
    raw_data = message.model_dump_json()
    # 3. 持久化（可选）
    await message_log_store.append(message.__class__.__name__, raw_data)
    # 4. 路由并投递到对应智能体的输入队列
    for subscriber in find_subscribers(message):
        await subscriber.input_queue.put(message)

4.4 测试框架：让多智能体系统可测试

多智能体系统的集成测试 notoriously hard（ notoriously hard）。我们构建了一个测试框架，核心思想是 模拟（Mock）和隔离 。

• 智能体单元测试 ：可以轻松Mock掉 llm_client 和 statsd_client ，只测试智能体内部的业务逻辑。
• 工作流集成测试 ：使用“内存消息总线”和“模拟LLM”来运行整个工作流。我们可以预先设定好模拟LLM的回复，然后断言工作流最终的状态和输出消息是否符合预期。
• 契约测试 ：利用强类型消息，我们可以自动生成并验证智能体之间接口的兼容性，防止某个智能体输出的消息格式变更导致下游崩溃。

@pytest.mark.asyncio
async def test_financial_analysis_workflow_happy_path():
    # 1. 使用测试专用的运行时容器，注入模拟组件
    test_runtime = create_test_runtime()
    test_runtime.mocker.llm_client.set_response(“这是一个上升趋势。”)

    # 2. 启动工作流
    input_msg = DataExtractionMessage(task_id="test-1", raw_data=[...])
    result = await test_runtime.run_workflow(“financial_analysis”, input_msg)

    # 3. 断言最终输出
    assert isinstance(result, ReportGeneratedMessage)
    assert “上升趋势” in result.content
    # 4. 断言中间状态和调用次数
    assert test_runtime.mocker.llm_client.call_count == 2

5. 实战对比：新运行时 vs. 旧框架方案

为了直观展示自研运行时的价值，我们用一个具体的业务场景——“客户投诉智能处理流程”——来对比实现差异。

场景描述 ：系统收到一封客户投诉邮件，需要自动：1) 提取关键信息并分类；2) 查询客户历史订单；3) 根据订单和投诉内容生成初步回复草稿；4) 若涉及赔偿，需计算金额并交由合规智能体复核；5) 最终生成处理工单。

使用 CrewAI/AutoGen 的典型实现痛点：

1. 流程控制散落 ：需要在多个智能体的提示词里写“如果涉及赔偿，请调用XX工具”，逻辑分散。
2. 数据传递麻烦 ：订单查询结果（JSON）和赔偿计算金额（数字）需要“塞”进自然语言描述中传递给下一个智能体，下游需要“猜”和“解析”。
3. 状态跟踪难 ：想知道一个投诉处理到哪一步了，需要去翻数据库里自定义的状态表，或者解析杂乱的对话日志。
4. 测试困难 ：需要启动整个框架，并模拟LLM的多次回复，测试成本高。

使用 Orchestra 运行时的实现：

# workflow.yaml (部分)
workflow:
  name: complaint_handling
  steps:
    - agent: classifier
      output_type: ComplaintClassificationMessage
    - agent: order_lookup
      depends_on: [classifier]
      output_type: OrderHistoryMessage
    - agent: draft_generator
      depends_on: [classifier, order_lookup]
      output_type: DraftResponseMessage
    - agent: compensation_calculator
      condition: “{{ steps.classifier.output.involves_compensation }}” # Jinja2模板，引用强类型字段
      depends_on: [order_lookup]
      output_type: CompensationProposalMessage
    - agent: compliance_checker
      condition: “{{ steps.compensation_calculator }}” # 仅当上一步执行了才执行
      depends_on: [compensation_calculator]
      output_type: ComplianceApprovalMessage
    - agent: ticket_generator
      depends_on: [draft_generator, compliance_checker?] # 可选依赖
      output_type: SupportTicketMessage

优势对比：

维度	传统框架方案	Orchestra 运行时方案	收益
流程清晰度	逻辑藏在对话和提示词中	YAML/代码显式定义 ，DAG可视化，一目了然	可维护性大幅提升
数据流	非结构化文本，需手动解析	强类型消息 ，自动验证，IDE智能提示	开发效率高，错误率低
状态管理	需自行实现	运行时内置上下文 ，自动持久化与恢复	原生支持长任务、断点续跑
错误处理	框架提供有限支持	任务级重试、超时、死信队列	系统健壮性增强
可观测性	控制台日志，需自行集成	内建Trace、Metrics、结构化日志	开箱即用的监控与调试
测试	复杂，需Mock整个对话链	分层测试框架 ，单元、集成测试方便	保障代码质量，加速迭代

6. 总结与反思：什么情况下你才需要自研？

经过半年的开发与线上运行，Orchestra运行时稳定支撑了我们核心的智能分析业务，吞吐量和可靠性均达到预期。回顾整个过程，我认为这次“造轮子”是成功的，但它并非适用于所有团队和场景。

在以下情况下，我强烈建议优先使用成熟框架（如LangGraph）：

• 你正在快速原型验证（PoC）或探索阶段 ：你的目标是快速验证AI智能体能否解决某个问题，而不是构建一个生产系统。成熟框架能让你在几小时内搭建出可运行的Demo。
• 你的团队AI工程经验不足 ：成熟框架提供了“最佳实践”的雏形，可以帮你避开许多初级陷阱，是一个很好的学习工具。
• 你的业务逻辑相对简单 ：主要是线性或简单的对话式任务，没有复杂的条件分支、状态管理或高性能要求。

而在以下情况下，你可以慎重考虑自研或深度定制：

• 你有明确的、复杂的生产级需求 ：涉及复杂工作流、严格的SLA（服务等级协议）、企业级集成和运维标准。
• 现有框架的抽象与你的心智模型严重不符 ：你发现你总是在“对抗”框架，写的胶水代码比业务代码还多。
• 你对性能、可控性和可观测性有极高要求 ：你需要精细控制内存、并发，需要深度的链路追踪和业务指标。
• 你的团队具备足够的工程能力 ：不仅限于AI应用开发，还包括分布式系统、异步编程、API设计等方面的经验。

自研一个运行时是一项重大的技术投资。它带来了极高的灵活性和对系统的深度掌控，但也伴随着持续的维护成本和初期开发投入。我们的故事并非鼓励所有人都去造轮子，而是想揭示一个事实： 当AI应用从玩具走向生产力工具时，其底层的“运行时”需求会变得越来越接近传统的分布式业务系统——稳定、可控、可观测、易维护。 现有的多智能体框架正在飞速进化，也许不久后它们就能完美覆盖这些生产需求。但在此之前，理解这些需求本身，并知道如何用软件工程的方法去满足它们，是每一个AI应用开发者走向深水区的必修课。

最终，我们构建的不仅是一个运行时，更是一套应对复杂AI应用工程的理念和方法。它可能以Orchestra的形式存在，也可能融入你对其他框架的深度定制中。核心在于， 不要让框架限制你对问题本质的思考，而是让工具服务于你清晰的架构设计。 当你对智能体如何协作、状态如何流转、故障如何隔离有了清晰的认识后，无论是选用框架还是自研组件，都将是一个从容而自信的选择。