--- title: "LLM工作流设计：构建智能应用的标准流程" description: "学习如何设计和实现LLM工作流，从简单的链式调用到复杂的多步骤智能应用" tags: ["LLM", "工作流", "Workflow", "智能应用", "流程设计"] category: "llm" icon: "🔄"

LLM工作流设计：构建智能应用的标准流程

什么是LLM工作流

LLM工作流（Workflow）是将大语言模型的多个处理步骤组织成自动化、可重复执行流程的设计模式。通过定义清晰的输入、处理和输出环节，我们可以构建可靠的智能应用。

基础工作流模式

1. 链式调用（Chain）

最简单的工作流模式是将多个LLM调用串联起来。

from openai import OpenAI

client = OpenAI()

class SimpleChain:
    def __init__(self):
        self.steps = []
    
    def add_step(self, name, prompt_template):
        self.steps.append({"name": name, "template": prompt_template})
        return self
    
    def execute(self, initial_input):
        current = initial_input
        for step in self.steps:
            response = client.chat.completions.create(
                model="gpt-4",
                messages=[{"role": "user", "content": step["template"].format(input=current)}]
            )
            current = response.choices[0].message.content
            print(f"[{step['name']}] 输出: {current[:100]}...")
        return current

# 使用示例
chain = SimpleChain()
chain.add_step("提取关键词", "从以下文本中提取5个关键词：{input}")
chain.add_step("生成摘要", "根据这些关键词生成一段摘要：{input}")

result = chain.execute("大语言模型正在改变软件开发的方式，从代码生成到测试自动化，AI助手已经成为开发者的得力工具。")

2. 路由工作流（Router）

根据输入内容动态选择不同的处理路径。

class RouterWorkflow:
    def __init__(self):
        self.routes = {}
        self.default_handler = None
    
    def register_route(self, category, handler):
        self.routes[category] = handler
    
    def classify(self, query):
        response = client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[{
                "role": "user",
                "content": f"将以下问题分类为：技术、业务、其他。只返回分类名称。\n\n{query}"
            }]
        )
        return response.choices[0].message.content.strip()
    
    def execute(self, query):
        category = self.classify(query)
        print(f"分类结果: {category}")
        
        handler = self.routes.get(category, self.default_handler)
        if handler:
            return handler(query)
        return "无法处理该请求"

# 定义处理器
def handle_technical(query):
    return f"技术回答: {query}"

def handle_business(query):
    return f"业务回答: {query}"

router = RouterWorkflow()
router.register_route("技术", handle_technical)
router.register_route("业务", handle_business)

3. 并行执行（Parallel）

同时执行多个独立的LLM任务。

import concurrent.futures

class ParallelWorkflow:
    def __init__(self):
        self.tasks = []
    
    def add_task(self, name, prompt):
        self.tasks.append({"name": name, "prompt": prompt})
    
    def execute_task(self, task):
        response = client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": task["prompt"]}]
        )
        return task["name"], response.choices[0].message.content
    
    def execute(self):
        with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
            futures = [executor.submit(self.execute_task, t) for t in self.tasks]
            results = {}
            for future in concurrent.futures.as_completed(futures):
                name, result = future.result()
                results[name] = result
        return results

# 使用示例
workflow = ParallelWorkflow()
workflow.add_task("翻译", "将'Hello World'翻译成中文、日文、韩文")
workflow.add_task("解释", "解释Hello World在编程中的意义")
workflow.add_task("历史", "介绍Hello World的历史起源")

results = workflow.execute()

复杂工作流设计

投票机制

class VotingWorkflow:
    def __init__(self, n_voters=3):
        self.n_voters = n_voters
    
    def get_vote(self, query, role):
        response = client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "system", "content": f"你是{role}"},
                     {"role": "user", "content": query}]
        )
        return response.choices[0].message.content
    
    def execute(self, query):
        roles = ["乐观分析师", "悲观分析师", "中立分析师"]
        votes = []
        for role in roles[:self.n_voters]:
            vote = self.get_vote(query, role)
            votes.append({"role": role, "opinion": vote})
        
        # 汇总投票
        summary_prompt = f"综合以下观点给出最终建议：\n" + "\n".join(
            [f"{v['role']}: {v['opinion']}" for v in votes]
        )
        final = client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": summary_prompt}]
        )
        return final.choices[0].message.content, votes

最佳实践

1. 错误处理：每个步骤都应包含重试和降级逻辑
2. 日志记录：记录每个步骤的输入输出，便于调试
3. 超时控制：为LLM调用设置合理的超时时间
4. 缓存机制：对相同输入的结果进行缓存
5. 监控告警：监控工作流执行状态和性能指标

总结

LLM工作流设计是构建可靠智能应用的基础。通过链式调用、路由分发、并行执行等模式，可以灵活组合出满足各种业务需求的工作流。关键是要做好错误处理、日志记录和监控，确保工作流的稳定运行。