AI编排：这次怎么落地的

从单一的模型调用到复杂的工作流编排，中间到底跨越了哪些坑，以及怎么才能尽量少踩一点。

错误处理：模型调用不是百分之百可靠的，可能会超时、失败、返回格式错误。

背景：为什么需要编排

最开始我们的 AI 能力很单一，就是一个搜索功能加一个摘要功能，各自调各自的模型接口。用户问"帮我找一下关于 xxx 的资料"，我们调搜索模型；用户说"把这个文档总结一下"，我们调摘要模型。这种模式在简单场景下没问题，但很快就遇到了几个现实问题。

一是用户意图越来越复杂。有用户开始问"帮我找一下关于 xxx 的资料，然后总结成要点，再根据这些要点给出一个实施方案"，这种跨多个环节的需求，单一模型调用搞不定。

二是业务逻辑越来越绕。有些需求需要先判断文档类型，再决定走哪条路径；有些需要并发调用多个接口，再汇总结果；还有些需要根据前一步的结果动态决定下一步做什么。这些都不是简单调用一个模型就能解决的。

三是维护成本开始上升。每个功能的逻辑散落在各个服务里，改一个点可能要动好几个地方，出错也不好定位。

我们开始意识到：需要一种机制，把这些 AI 能力有机地组织起来，既能处理复杂意图，又能保持代码的可维护性。这就是 AI 编排的起点。

需求：要编排些什么

在实际业务场景中，AI 编排主要解决这几个问题：

状态流转：一个任务从开始到结束，中间会经历多个步骤，每个步骤的输出是下一步的输入。比如用户上传文档，先要做 OCR，再做摘要，再做问答，最后还要打标签。这些步骤之间有明确的依赖关系，需要按顺序执行。

分支决策：不是所有任务都走同样的路径。比如 PDF 文档和 Word 文档可能需要不同的预处理策略；长文档和短文档可能需要不同的摘要模型；有明确问题和模糊问题的处理方式也不一样。这些分支判断需要写在工作流里。

并发控制：有些步骤可以并行执行，比如对一个长文档，可以同时做摘要、打标签、提取关键词。但不是所有并发都是好事，模型调用有成本，而且 API 也有速率限制。需要合理控制并发度，在速度和成本之间找平衡。

错误处理：模型调用不是百分之百可靠的，可能会超时、失败、返回格式错误。需要设计合理的重试策略、降级方案和错误上报机制。

成本优化：每个模型调用都有成本，能不用就不用，能用小的就不用大的。比如简单的分类任务，小模型就够了，不需要每次都上大模型。

这些问题看起来都不复杂，但组合在一起就是一个工程挑战。我们一开始自己写了一个简单的编排框架，后来发现重复造轮子不如直接用成熟的方案。

实现：从自研到 LangChain

早期方案：自研编排框架

最开始我们尝试自己写一个编排框架，核心思路是把每个 AI 操作抽象成一个节点，节点之间通过数据流连接。

这个方案的优点是完全可控，可以针对我们的业务场景做深度定制。但很快发现有几个问题：

一是维护成本高。每次新增一个功能，都要改框架代码，慢慢就变得复杂难以维护。

二是通用性差。换一个业务场景，可能要大改框架代码，复用性低。

三是缺少生态。很多常见功能，比如记忆管理、链路追踪、工具调用，都要自己实现，重复造轮子。

用了一段时间后，我们决定看看业界有没有成熟的方案。

技术选型：LangChain 的评估

调研了一圈，主要看了 LangChain、Semantic Kernel、Haystack 这些框架。最终选择 LangChain 主要是这几个原因：

生态成熟：LangChain 有丰富的集成，支持各种模型、工具、向量数据库，社区也比较活跃。

灵活性好：它不是一个黑盒框架，而是一系列可组合的组件，可以根据需要自由组合。

文档清晰：虽然有段时间文档质量受诟病，但整体上还是比较好上手的。

但也要说清楚，LangChain 不是完美的。它的学习曲线比较陡，概念比较多，初学者容易迷路。而且版本迭代快，API 经常变化，升级时要小心。

架构设计：分层编排

基于 LangChain，我们设计了一个分层架构。底层是模型层，封装各种模型调用；中间是编排层，用 Chains 和 Agents 组织逻辑；上层是业务层，处理用户请求和响应。

这个架构的好处是分层清晰，每层各司其职。业务层只负责用户交互，编排层负责逻辑组织，模型层负责模型调用。

核心实现：Chains 和 Agents

在 LangChain 里，Chains 用于处理确定性流程，Agents 用于处理动态决策。

简单链：最基础的链，一个步骤接一个步骤。比如先做 OCR，再做摘要。

from langchain.chains import SimpleSequentialChain

ocr_chain = OCROverflowChain()
summary_chain = SummaryChain()

chain = SimpleSequentialChain(chains=[ocr_chain, summary_chain])

条件链：根据条件走不同分支。比如根据文档类型选择不同的处理方式。

from langchain.chains import ConditionalBranchChain

branch_chain = ConditionalBranchChain(
    branches={
        "pdf": pdf_processing_chain,
        "word": word_processing_chain,
        "default": default_processing_chain
    },
    condition=lambda x: x["file_type"]
)

Agent：让模型自己决定用什么工具。比如用户问"帮我查一下天气"，Agent 可以自动选择调用天气 API。

from langchain.agents import initialize_agent, Tool

tools = [
    Tool(name="Weather", func=weather_api),
    Tool(name="Search", func=search_api)
]

agent = initialize_agent(
    tools, llm, agent="zero-shot-react-description"
)

在实际项目中，我们会把常用逻辑封装成 Chains，需要动态决策的地方用 Agents，两者配合使用。

踩坑：几个关键问题

实现过程中遇到了不少坑，这里挑几个比较有代表性的说。

状态管理的坑

一开始我们用全局变量维护状态，很快发现不行。并发请求会互相干扰，状态也会混乱。

后来改用 LangChain 的 Memory 组件，但发现 Memory 主要用于对话场景，不太适合工作流状态。

最终我们采用了一个折中方案：用 Redis 存储工作流状态，每个节点从 Redis 读状态、写状态。这样既保证了并发安全，又能持久化状态。

import redis

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def process_step(task_id, step_name, process_func):
    state = json.loads(redis_client.get(f"task:{task_id}"))
    result = process_func(state)
    state[step_name] = result
    redis_client.set(f"task:{task_id}", json.dumps(state))
    return result

错误重试的坑

模型调用失败是常态，但重试策略不好设计。重试太频繁可能被限流，重试太少可能导致任务失败。

我们采用了指数退避策略，同时设置了最大重试次数和超时时间。

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10))
def call_model_with_retry(prompt):
    return llm.predict(prompt)

但即使这样，还是有些边界情况没考虑到，比如某些错误不应该重试（比如参数错误），有些错误应该特殊处理（比如配额不足）。这些都需要根据实际场景细化。

成本控制的坑

一开始没注意成本，发现账单时已经晚了。大模型调用的成本不容忽视。

我们后来做了几个优化：

一是模型选择。简单任务用小模型，复杂任务才用大模型。比如分类任务用较小的模型，摘要任务用大一点的模型。

二是缓存。对相同的输入，缓存结果，避免重复调用。

三是 Prompt 优化。精简 Prompt，减少 token 消耗。

from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_summarize(text):
    return summarization_chain.run(text)

这些优化能显著降低成本，但也要注意缓存的时效性和清理策略。

结果：效果和经验

折腾了一圈，最终的效果还是不错的。我们实现了：

意图理解准确率提升：从单一模型调用的 60% 左右提升到编排后的 85% 左右。主要是通过多步骤的意图识别和上下文理解。

处理速度优化：通过并发控制和合理调度，整体响应时间从平均 15 秒降低到 8 秒左右。

成本降低：通过模型选择、缓存和 Prompt 优化，成本降低了约 40%。

可维护性提升：业务逻辑和编排逻辑分离，代码结构清晰，新增功能容易。

也有一些经验教训：

不要过度编排：不是所有场景都需要编排，简单任务直接调用模型可能更高效。

先想清楚再动手：编排逻辑越复杂，调试成本越高，设计阶段多花时间能省很多麻烦。

监控很重要：没有监控就不知道哪里出了问题，我们后来加上了链路追踪和性能监控，问题定位容易多了。

结语

AI 编排不是银弹，它解决的是如何把多个 AI 能力有机组织起来的问题。但实际问题往往比想象的复杂，需要权衡很多因素：性能、成本、可维护性、开发效率……

这次实践让我们明白：技术选型要看场景，框架好不好用要看适不适合。LangChain 解决了我们的问题，但不一定解决别人的问题。

写这篇文章的时候，AI 编排这个领域还在快速变化，新的框架、新的工具层出不穷。但底层的问题其实没变：如何组织复杂的逻辑，如何处理不确定性，如何在成本和效果之间找平衡。

这些问题，估计还会继续困扰我们一段时间。

版权声明: 本文首发于 指尖魔法屋-AI编排：这次怎么落地的（https://blog.thinkmoon.cn/post/266-ai-orchestration-model-workflow-practice/） 转载或引用必须申明原指尖魔法屋来源及源地址！