anthropics-claude-cookbooks学习记录01

---

学习记录claude-cookbooks。
今日学习：https://github.com/anthropics/claude-cookbooks/tree/main/patterns/agents

basic_workflows.ipynb

Basic Multi-LLM Workflows

本笔记本展示了三种简单的多大语言模型（Multi-LLM）工作流。它们在成本或延迟方面做出权衡，以期获得潜在的任务性能提升：

• 提示链（Prompt-Chaining）：将任务分解为一系列顺序执行的子任务，每一步都基于前一步的结果进行构建。
• 并行化（Parallelization）：将相互独立的子任务分发给多个大语言模型，实现并发处理。
• 路由（Routing）：根据输入特征动态选择专门优化的大语言模型路径。

注意：这些是用于演示核心概念的示例实现，并非生产级代码。

from concurrent.futures import ThreadPoolExecutor
from typing import List, Dict, Callable
from util import llm_call, extract_xmldef chain(input: str, prompts: List[str]) -> str:"""依次串联多次大语言模型（LLM）调用，将前一步的结果作为下一步的输入。"""result = inputfor i, prompt in enumerate(prompts, 1):print(f"\n步骤 {i}:")result = llm_call(f"{prompt}\n输入: {result}")print(result)return resultdef parallel(prompt: str, inputs: List[str], n_workers: int = 3) -> List[str]:"""使用相同的提示（prompt），并发处理多个输入。"""with ThreadPoolExecutor(max_workers=n_workers) as executor:# 为每个输入提交一个异步 LLM 调用任务futures = [executor.submit(llm_call, f"{prompt}\n输入: {x}") for x in inputs]# 等待所有任务完成并收集结果return [f.result() for f in futures]def route(input: str, routes: Dict[str, str]) -> str:"""根据输入内容自动选择最合适的专用处理路径（prompt）。"""# 第一步：使用大语言模型对输入进行分类，决定应路由到哪个路径print(f"\n可用路由: {list(routes.keys())}")selector_prompt = f"""请分析以下输入，并从下列选项中选择最合适的支持团队：{list(routes.keys())}请先说明你的推理过程，然后在以下 XML 格式中给出你的选择：<reasoning>简要解释为何该工单应分配给某个特定团队。请考虑关键词、用户意图和紧急程度等因素。</reasoning><selection>所选团队名称</selection>输入: {input}""".strip()# 调用 LLM 进行路由决策route_response = llm_call(selector_prompt)reasoning = extract_xml(route_response, 'reasoning')      # 提取推理理由route_key = extract_xml(route_response, 'selection').strip().lower()  # 提取所选路径名称（转为小写）print("路由分析:")print(reasoning)print(f"\n选定路由: {route_key}")# 使用选定的专用提示处理原始输入selected_prompt = routes[route_key]return llm_call(f"{selected_prompt}\n输入: {input}")

Example Usage

以下是每种工作流的实用示例说明：

1. 链式工作流（Chain Workflow）：用于结构化数据提取与格式化
2. 并行化工作流（Parallelization Workflow）：用于利益相关方影响分析
3. 路由工作流（Route Workflow）：用于客服工单处理

# Example 1: Chain workflow for structured data extraction and formatting
# Each step progressively transforms raw text into a formatted tabledata_processing_steps = ["""从文本中仅提取数值及其关联指标。每条数据以'数值: 指标'的格式换行显示。示例格式：92: 客户满意度45%: 收入增长率""","""将所有数值尽可能转换为百分比形式。若非百分比或点数，则转换为小数（例如：92 points → 92%）。每行保留一个数值。示例格式：92%: 客户满意度45%: 收入增长率""","""按数值从高到低对所有行进行降序排序。保持每行'数值: 指标'的格式。示例：92%: 客户满意度87%: 员工满意度""","""将排序后的数据格式化为Markdown表格，包含列：| 指标 | 数值 ||:--|--:|| 客户满意度 | 92% |"""
]report = """
第三季度绩效总结：
我们的客户满意度得分本季度升至92 points。
收入较去年增长45%。
主要市场份额现已达到23%。
客户流失率从8%降至5%。
新用户获取成本为每用户43美元。
产品使用率提升至78%。
员工满意度为87分。
营业利润率改善至34%。
"""print("\n输入文本：")
print(report)
formatted_result = chain(report, data_processing_steps)

# Example 2: Parallelization workflow for stakeholder impact analysis
# Process impact analysis for multiple stakeholder groups concurrentlystakeholders = ["""客户：- 对价格敏感- 希望更好的技术- 关注环境问题""","""员工：- 担心工作保障- 需要新技能- 希望明确方向""","""投资者：- 期望增长- 希望控制成本- 关注风险""","""供应商：- 产能限制- 价格压力- 技术转型"""
]impact_results = parallel("""分析市场变化将如何影响此利益相关方群体。提供具体影响和推荐行动。格式要求清晰的章节和优先级。""",stakeholders
)for result in impact_results:print(result)print('+' * 80)

# Example 3: Route workflow for customer support ticket handling
# Route support tickets to appropriate teams based on content analysissupport_routes = {"billing": """您是一名计费服务专家。请遵循以下指南：1. 始终以"计费支持回复："开头2. 首先确认具体的计费问题3. 清晰地解释任何费用或差异4. 列出具体的后续步骤和时间安排5. 如相关，以付款选项结尾保持回复专业但友好。输入：""","technical": """您是一名技术支持工程师。请遵循以下指南：1. 始终以"技术支持回复："开头2. 列出解决问题的确切步骤3. 如相关，包含系统要求4. 提供常见问题的临时解决方案5. 如需要，提供升级路径使用清晰的编号步骤和技术细节。输入：""","account": """您是一名账户安全专家。请遵循以下指南：1. 始终以"账户支持回复："开头2. 优先处理账户安全和验证3. 提供账户恢复/更改的清晰步骤4. 包含安全提示和警告5. 设定明确的解决时间预期保持严肃、以安全为重点的语气。输入：""","product": """您是一名产品专家。请遵循以下指南：1. 始终以"产品支持回复："开头2. 专注于功能教育和最佳实践3. 包含具体的使用示例4. 链接到相关文档部分5. 建议可能有帮助的相关功能语气要有教育性和鼓励性。输入："""
}# Test with different support tickets
tickets = ["""主题：无法访问我的账户消息：您好，我过去一小时一直在尝试登录，但一直收到"密码错误"的提示。我确定我使用的是正确的密码。您能帮我重新获得访问权限吗？这很紧急，因为我需要在今天下班前提交报告。- John""","""主题：信用卡上的意外收费消息：您好，我刚刚注意到我的信用卡上有您公司的一笔49.99美元收费，但我以为我使用的是29.99美元的计划。您能解释这笔费用吗？如果是错误，请进行调整。谢谢，Sarah""","""主题：如何导出数据？消息：我需要将所有项目数据导出到Excel。我查阅了文档，但不知道如何进行批量导出。这可能吗？如果可以，您能带我一步步完成吗？诚挚问候，Mike"""
]print("正在处理支持工单...\n")
for i, ticket in enumerate(tickets, 1):print(f"\n工单 {i}:")print("-" * 40)print(ticket)print("\n回复：")print("-" * 40)response = route(ticket, support_routes)print(response)print("+" * 80)

evaluator_optimizer.ipynb

Evaluator-Optimizer Workflow

该工作流通过双模型协作实现：一个LLM生成回答，另一个则通过循环机制进行评估与反馈。

该工作流在以下场景中尤为有效：

• 具备明确的评估标准
• 需要通过迭代优化提升价值
  其适用性主要体现在两个特征：
• 当提供反馈后，LLM的响应质量能够获得显著提升
• LLM自身能够提供具有实质意义的反馈建议

from util import llm_call, extract_xmldef generate(prompt: str, task: str, context: str = "") -> tuple[str, str]:"""生成解决方案并根据反馈进行改进"""full_prompt = f"{prompt}\n{context}\n任务: {task}" if context else f"{prompt}\n任务: {task}"response = llm_call(full_prompt)thoughts = extract_xml(response, "thoughts")result = extract_xml(response, "response")print("\n=== 生成开始 ===")print(f"思考过程:\n{thoughts}\n")print(f"生成结果:\n{result}")print("=== 生成结束 ===\n")return thoughts, resultdef evaluate(prompt: str, content: str, task: str) -> tuple[str, str]:"""评估解决方案是否满足要求"""full_prompt = f"{prompt}\n原始任务: {task}\n待评估内容: {content}"response = llm_call(full_prompt)evaluation = extract_xml(response, "evaluation")feedback = extract_xml(response, "feedback")print("=== 评估开始 ===")print(f"评估状态: {evaluation}")print(f"反馈意见: {feedback}")print("=== 评估结束 ===\n")return evaluation, feedbackdef loop(task: str, evaluator_prompt: str, generator_prompt: str) -> tuple[str, list[dict]]:"""持续生成和评估，直到满足要求为止"""memory = []chain_of_thought = []thoughts, result = generate(generator_prompt, task)memory.append(result)chain_of_thought.append({"thoughts": thoughts, "result": result})while True:evaluation, feedback = evaluate(evaluator_prompt, result, task)if evaluation == "PASS":return result, chain_of_thoughtcontext = "\n".join(["先前尝试:",*[f"- {m}" for m in memory],f"\n反馈意见: {feedback}"])thoughts, result = generate(generator_prompt, task, context)memory.append(result)chain_of_thought.append({"thoughts": thoughts, "result": result})

应用场景示例：迭代式代码开发循环：

evaluator_prompt = """
请对以下代码进行评估：
1. 代码正确性
2. 时间复杂度
3. 代码风格和最佳实践你只应该进行评估，不要尝试解决任务。
只有当所有标准都满足且没有改进建议时，才输出"PASS"。
请按以下格式简洁地输出评估结果。<evaluation>通过, 需要改进, 或不通过</evaluation>
<feedback>
需要改进的内容及原因。
</feedback>
"""generator_prompt = """
你的目标是根据<用户输入>完成任务。如果之前生成的结果有反馈意见，
你应该反思这些反馈以改进你的解决方案。请按以下格式简洁地输出答案：<thoughts>
[你对任务和反馈的理解，以及你计划如何改进]
</thoughts><response>
[你的代码实现放在这里]
</response>
"""task = """
<用户输入>
实现一个栈，包含以下功能：
1. push(x) - 入栈
2. pop() - 出栈  
3. getMin() - 获取最小值
所有操作都应该是O(1)时间复杂度。
</用户输入>
"""loop(task, evaluator_prompt, generator_prompt)

orchestrator_workers.ipynb

Orchestrator-Workers Workflow

协调者-工作者工作流
在此工作流中，一个作为“协调者”的核心LLM会动态解析总任务，将其分派给多个“工作者”LLM去执行，并最终整合各工作者的输出结果。

适用场景
该工作流尤其适用于任务复杂、无法预先确定所有子任务的场景。其与简单并行处理的核心区别在于灵活性——子任务并非固定不变，而是由协调者根据当前输入动态生成。

from typing import Dict, List, Optional
from util import llm_call, extract_xmldef parse_tasks(tasks_xml: str) -> List[Dict]:"""解析XML格式的任务列表，返回任务字典列表"""tasks = []current_task = {}for line in tasks_xml.split('\n'):line = line.strip()if not line:continueif line.startswith("<task>"):current_task = {}elif line.startswith("<type>"):current_task["type"] = line[6:-7].strip()elif line.startswith("<description>"):current_task["description"] = line[12:-13].strip()elif line.startswith("</task>"):if "description" in current_task:if "type" not in current_task:current_task["type"] = "default"  # 默认类型tasks.append(current_task)return tasksclass FlexibleOrchestrator:"""动态分解任务并使用工作者LLM并行执行的协调器"""def __init__(self,orchestrator_prompt: str,worker_prompt: str,):"""使用提示模板初始化协调器"""self.orchestrator_prompt = orchestrator_promptself.worker_prompt = worker_promptdef _format_prompt(self, template: str, **kwargs) -> str:"""使用变量格式化提示模板"""try:return template.format(**kwargs)except KeyError as e:raise ValueError(f"缺少必要的提示变量: {e}")def process(self, task: str, context: Optional[Dict] = None) -> Dict:"""处理任务：分解任务并并行执行子任务"""context = context or {}# 第一步：获取协调器响应orchestrator_input = self._format_prompt(self.orchestrator_prompt,task=task,**context)orchestrator_response = llm_call(orchestrator_input)# 解析协调器响应analysis = extract_xml(orchestrator_response, "analysis")tasks_xml = extract_xml(orchestrator_response, "tasks")tasks = parse_tasks(tasks_xml)print("\n=== 协调器输出 ===")print(f"\n分析结果:\n{analysis}")print(f"\n任务列表:\n{tasks}")# 第二步：处理每个子任务worker_results = []for task_info in tasks:worker_input = self._format_prompt(self.worker_prompt,original_task=task,task_type=task_info['type'],task_description=task_info['description'],**context)worker_response = llm_call(worker_input)result = extract_xml(worker_response, "response")worker_results.append({"type": task_info["type"],"description": task_info["description"],"result": result})print(f"\n=== 工作者结果 ({task_info['type']}) ===\n{result}\n")return {"analysis": analysis,"worker_results": worker_results,}

示例应用场景：营销素材多版本生成

ORCHESTRATOR_PROMPT = """
分析以下任务并将其分解为2-3种不同的创作方向：任务：{task}请按以下格式返回响应：<analysis>
阐述你对任务的理解，并说明每种创作方向的价值所在。
重点说明每种方法如何服务于任务的不同方面。
</analysis><tasks><task><type>正式版</type><description>撰写精确、专业的版本，重点突出产品规格参数</description></task><task><type>口语版</type><description>撰写生动有趣的版本，与读者建立情感连接</description></task>
</tasks>
"""WORKER_PROMPT = """
根据以下要求生成内容：
原始任务：{original_task}
文案风格：{task_type}
创作指南：{task_description}请按以下格式返回响应：<response>
请在此处填写生成的内容，保持指定风格并完整满足所有要求。
</response>
"""orchestrator = FlexibleOrchestrator(orchestrator_prompt=ORCHESTRATOR_PROMPT,worker_prompt=WORKER_PROMPT,
)results = orchestrator.process(task="为新型环保水壶撰写产品描述",context={"target_audience": "注重环保的千禧一代","key_features": ["无塑料材质", "保温功能", "终身保修"]}
)