llmintegration

LLM 集成技能

为 Golden Armada AI Agent Fleet Platform 提供全面的 LLM 集成能力，包括扩展思考、高级提示词工程和智能令牌预算管理等高级功能。

何时使用此技能

在以下场景中使用此技能：

• Claude/Anthropic API 集成
• OpenAI GPT 集成
• Google Gemini 集成
• Ollama 本地模型
• 多模型编排
• 提示词工程

Anthropic Claude 集成

```python import anthropic

client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

基本补全

message = client.messages.create( model="claude-sonnet-5", max_tokens=1024, messages=[ {"role": "user", "content": "你好，Claude！"} ] ) print(message.content[0].text)

使用系统提示词

message = client.messages.create( model="claude-sonnet-5", max_tokens=1024, system="你是一名乐于助人的编码助手。", messages=[ {"role": "user", "content": "编写一个用于对列表排序的 Python 函数。"} ] )

流式传输

with client.messages.stream( model="claude-sonnet-5", max_tokens=1024, messages=[{"role": "user", "content": "给我讲个故事。"}] ) as stream: for text in stream.text_stream: print(text, end="", flush=True)

工具调用

tools = [ { "name": "get_weather", "description": "获取某个地点的当前天气", "input_schema": { "type": "object", "properties": { "location": {"type": "string", "description": "城市和州"} }, "required": ["location"] } } ]

message = client.messages.create( model="claude-sonnet-5", max_tokens=1024, tools=tools, messages=[{"role": "user", "content": "旧金山的天气怎么样？"}] ) ```

OpenAI GPT 集成

```python from openai import OpenAI

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

基本补全

response = client.chat.completions.create( model="gpt-4", messages=[ {"role": "system", "content": "你是一名乐于助人的助手。"}, {"role": "user", "content": "你好！"} ] ) print(response.choices[0].message.content)

流式传输

stream = client.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": "写一首诗。"}], stream=True ) for chunk in stream: if chunk.choices[0].delta.content: print(chunk.choices[0].delta.content, end="")

函数调用

functions = [ { "name": "get_weather", "description": "获取当前天气", "parameters": { "type": "object", "properties": { "location": {"type": "string"} }, "required": ["location"] } } ]

response = client.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": "纽约市的天气怎么样？"}], functions=functions, function_call="auto" ) ```

Google Gemini 集成

```python import google.generativeai as genai

genai.configure(api_key=os.environ["GOOGLE_API_KEY"])

model = genai.GenerativeModel('gemini-pro')

基本生成

response = model.generate_content("解释量子计算") print(response.text)

聊天

chat = model.start_chat(history=[]) response = chat.send_message("你好！") print(response.text)

流式传输

response = model.generate_content("给我讲个故事", stream=True) for chunk in response: print(chunk.text, end="") ```

Ollama 本地模型

```python import ollama

基本补全

response = ollama.chat( model='llama2', messages=[ {'role': 'user', 'content': '你好！'} ] ) print(response['message']['content'])

流式传输

stream = ollama.chat( model='llama2', messages=[{'role': 'user', 'content': '给我讲个故事。'}], stream=True ) for chunk in stream: print(chunk['message']['content'], end='')

拉取模型

ollama.pull('llama2')

列出模型

models = ollama.list() ```

多模型抽象

```python from abc import ABC, abstractmethod from typing import Generator

class LLMProvider(ABC): @abstractmethod def generate(self, prompt: str, **kwargs) -> str: pass

@abstractmethod
def stream(self, prompt: str, **kwargs) -> Generator[str, None, None]:
    pass

class ClaudeProvider(LLMProvider): def init(self, api_key: str, model: str = "claude-sonnet-5"): self.client = anthropic.Anthropic(api_key=api_key) self.model = model

def generate(self, prompt: str, **kwargs) -> str:
    message = self.client.messages.create(
        model=self.model,
        max_tokens=kwargs.get('max_tokens', 1024),
        messages=[{"role": "user", "content": prompt}]
    )
    return message.content[0].text

def stream(self, prompt: str, **kwargs) -> Generator[str, None, None]:
    with self.client.messages.stream(
        model=self.model,
        max_tokens=kwargs.get('max_tokens', 1024),
        messages=[{"role": "user", "content": prompt}]
    ) as stream:
        for text in stream.text_stream:
            yield text

class LLMFactory: @staticmethod def create(provider: str, **kwargs) -> LLMProvider: providers = { 'claude': ClaudeProvider, 'gpt': GPTProvider, 'gemini': GeminiProvider, 'ollama': OllamaProvider } return providersprovider ```

提示词工程最佳实践

```python

结构化提示词

SYSTEM_PROMPT = """你是一位乐于助人的编码助手。

指南：

1. 编写简洁、文档完善的代码
2. 遵循最佳实践
3. 解释你的推理过程 """

少样本示例

FEW_SHOT_PROMPT = """将自然语言转换为 SQL。

示例 1： 输入：获取所有用户 输出：SELECT * FROM users;

示例 2： 输入：统计活跃订单 输出：SELECT COUNT(*) FROM orders WHERE status = 'active';

输入：{user_input} 输出："""

思维链

COT_PROMPT = """请逐步解决这个问题： {problem}

让我们一步步思考： 1.""" ```

扩展思考集成

扩展思考使 Claude 模型能够在回复前进行“思考”，从而提高其在编码、数学和科学推理等复杂任务上的准确性。

何时使用扩展思考

• 复杂的多步骤推理任务
• 代码架构和系统设计
• 数学问题求解
• 科学分析和研究
• 战略规划和决策

交叉参考： 请参阅 .claude/skills/extended-thinking/SKILL.md，了解详细指南。

基本扩展思考

```python import anthropic

client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

自适应思考（当前模型：Opus 4.6+、Sonnet 5、Fable 5）

注意：Opus 4.7+/Sonnet 5/Fable 5 已移除 budget_tokens（会返回 400）；

使用 output_config 的 effort 等级来控制思考深度

response = client.messages.create( model="claude-sonnet-5", max_tokens=16000, thinking={"type": "adaptive"}, output_config={"effort": "high"}, messages=[ { "role": "user", "content": "为多租户 SaaS 平台设计可扩展的微服务架构" } ] )

响应包含思考内容和最终文本

for block in response.content: if block.type == "thinking": print(f"思考：{block.thinking}") elif block.type == "text": print(f"回复：{block.text}") ```

使用流式传输进行扩展思考

```python

实时流式输出思考过程

with client.messages.stream( model="claude-sonnet-5", max_tokens=16000, thinking={"type": "adaptive"}, messages=[ {"role": "user", "content": "分析此排序算法的时间复杂度..."} ] ) as stream: for event in stream: if event.type == "content_block_start": if event.content_block.type == "thinking": print("\n[思考过程]") elif event.content_block.type == "text": print("\n[最终答案]") elif event.type == "content_block_delta": if event.delta.type == "thinking_delta": print(event.delta.thinking, end="", flush=True) elif event.delta.type == "text_delta": print(event.delta.text, end="", flush=True) ```

多提供商扩展思考抽象

```python from typing import Optional, Dict, Any from dataclasses import dataclass

@dataclass class ThinkingConfig: enabled: bool = False effort: str = "high" # low | medium | high | xhigh | max show_thinking: bool = True

class ExtendedThinkingProvider: """跨提供商的扩展思考抽象"""

def __init__(self, provider: str, config: ThinkingConfig):
    self.provider = provider
    self.config = config

def generate_with_thinking(self, prompt: str, **kwargs) -> Dict[str, Any]:
    if self.provider == "claude":
        return self._claude_thinking(prompt, **kwargs)
    elif self.provider == "gpt":
        # 使用思维链模拟思考
        return self._gpt_cot_thinking(prompt, **kwargs)
    else:
        raise ValueError(f"提供商 {self.provider} 不支持扩展思考")

def _claude_thinking(self, prompt: str, **kwargs) -> Dict[str, Any]:
    client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

    thinking_params = {}
    if self.config.enabled:
        thinking_params["thinking"] = {"type": "adaptive"}
        thinking_params["output_config"] = {"effort": self.config.effort}

    response = client.messages.create(
        model=kwargs.get("model", "claude-sonnet-5"),
        max_tokens=kwargs.get("max_tokens", 16000),
        messages=[{"role": "user", "content": prompt}],
        **thinking_params
    )

    result = {"thinking": None, "response": None}
    for block in response.content:
        if block.type == "thinking":
            result["thinking"] = block.thinking
        elif block.type == "text":
            result["response"] = block.text

    return result

def _gpt_cot_thinking(self, prompt: str, **kwargs) -> Dict[str, Any]:
    """对 GPT 模型使用思维链提示词"""
    from openai import OpenAI
    client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

    cot_prompt = f"""我们按步骤来分析：

{prompt}

首先，系统地思考这个问题，然后给出最终答案。"""

    response = client.chat.completions.create(
        model=kwargs.get("model", "gpt-4"),
        messages=[{"role": "user", "content": cot_prompt}]
    )

    # 从回复中解析思考内容（基于启发式方法）
    content = response.choices[0].message.content
    parts = content.split("\n\n")

    return {
        "thinking": "\n\n".join(parts[:-1]) if len(parts) > 1 else None,
        "response": parts[-1] if parts else content
    }

用法

thinking_provider = ExtendedThinkingProvider( provider="claude", config=ThinkingConfig(enabled=True, effort="high") )

result = thinking_provider.generate_with_thinking( "为多租户系统设计分布式缓存策略" )

if result["thinking"]: print(f"思考过程：\n{result['thinking']}\n") print(f"最终答案：\n{result['response']}") ```

Claude 提示词工程最佳实践

基于 Anthropic 官方指南，以获得最佳性能。

1. 清晰、直接的指令

```python

不佳：请求含糊

prompt = "让这段代码更好"

良好：具体指令

prompt = """重构此 Python 函数，以便：

1. 使用类型提示
2. 添加全面的文档字符串
3. 处理边缘情况（空输入、None 值）
4. 改进变量命名，使其更清晰
5. 添加输入验证

代码： {code} """ ```

2. 使用 XML 标签组织结构

```python

XML 标签有助于 Claude 解析复杂输入

prompt = """分析此代码库并找出安全漏洞。

{auth_code} {api_code} - 重点检查身份验证和授权问题 - 检查 SQL 注入漏洞 - 找出不当的输入验证 - 标记缺少速率限制的情况

请按以下格式输出： High|Medium|Low path/to/file.py 123 清晰的描述 修复方法 """ ```

3. 提供示例（少样本提示）

```python FEW_SHOT_TEMPLATE = """将用户故事转换为验收标准。

作为一名用户，我想重置密码，以便重新访问我的账户。 - 假设我在登录页面 - 当我点击“忘记密码”时 - 那么我应该看到密码重置表单 - 并且我应该通过电子邮件收到重置链接 - 并且该链接应在 1 小时后过期 - 并且我可以通过该链接设置新密码 作为一名管理员，我想导出用户数据，以便生成报告。 - 假设我以管理员身份登录 - 当我导航到 Users > Export 时 - 那么我应该看到导出格式选项（CSV、JSON、Excel） - 并且我应该能够按日期范围筛选 - 并且导出内容应包含所有用户字段 - 并且我应该在 5 分钟内收到下载链接

现在转换此用户故事： <user_story> {user_input} </user_story> """ ```

4. 指定角色以提供上下文

```python ROLE_BASED_SYSTEM_PROMPTS = { "code_reviewer": """你是一位拥有 15 年经验的代码审查专家。 你的专长包括：

• 软件架构和设计模式
• 安全最佳实践（OWASP Top 10）
• 性能优化
• 代码可维护性和可读性

审查代码时：

1. 找出 Bug 和潜在问题
2. 针对清晰度和性能提出改进建议
3. 检查安全漏洞
4. 建议改进设计模式
5. 确保代码遵循语言最佳实践

反馈要有建设性且具体。""",

"architect": """你是一位资深软件架构师，专长包括：

• 微服务和分布式系统
• 云原生架构（AWS、GCP、Azure）
• 数据库设计和优化
• API 设计（REST、GraphQL、gRPC）
• 安全和合规

设计系统时：

1. 考虑可扩展性和性能

2. 确保容错性和弹性

3. 设计可观测性（日志、指标、追踪）

4. 遵循云原生最佳实践

5. 考虑成本优化""",

   "security_expert": """你是一位专注于以下领域的安全专家：

• OWASP Top 10 漏洞
• 身份验证和授权（OAuth、OIDC、JWT）
• 数据加密和隐私
• 安全编码实践
• 合规（GDPR、HIPAA、SOC2）

分析安全问题时：

1. 按严重程度标识漏洞
2. 提供具体的修复步骤
3. 参考安全标准和最佳实践
4. 同时考虑代码级和架构级安全""" }

用法

message = client.messages.create( model="claude-sonnet-5", max_tokens=4096, system=ROLE_BASED_SYSTEM_PROMPTS["code_reviewer"], messages=[ {"role": "user", "content": f"Review this code:\n\n{code}"} ] ) ```

5. 思维链提示

```python COT_PROMPT = """请逐步解决这个问题，并展示每个阶段的推理过程。

问题：{problem}

请按以下步骤思考：

1. 理解问题的要求
2. 找出相关信息
3. 将问题拆分为多个步骤
4. 逐步解决问题
5. 验证解决方案

展示每一步的推理过程。"""

对于复杂推理，可结合扩展思考

response = client.messages.create( model="claude-sonnet-5", max_tokens=16000, thinking={"type": "adaptive"}, messages=[ {"role": "user", "content": COT_PROMPT.format(problem=complex_problem)} ] ) ```

6. 用结构化输出控制格式

```python

注意：在当前模型中，助手轮次预填充会返回 400

（Opus 4.6+、Sonnet 5、Fable 5）。请改用结构化输出：

response = client.messages.create( model="claude-sonnet-5", max_tokens=1024, output_config={ "format": { "type": "json_schema", "schema": { "type": "object", "properties": { "entities": {"type": "array", "items": {"type": "string"}} }, "required": ["entities"], "additionalProperties": False } } }, messages=[ {"role": "user", "content": "Extract entities from: 'Apple Inc. hired John Smith as CEO in 2023.'"} ] )

输出保证符合架构

json_output = response.content[0].text ```

7. 长上下文最佳实践

```python

对于 Claude 的 1M token 上下文窗口（Haiku 4.5 为 200k）

LONG_CONTEXT_TEMPLATE = """我将提供一个大型代码库供分析。 与此任务最相关的文件位于此消息的末尾。

{less_important_files}

<critical_files> {important_files} </critical_files>

{specific_question}

回答任务时，请优先关注关键文件。""" ```

Token 预算管理

Anthropic Token 计数

```python import anthropic

def count_tokens_anthropic(text: str, model: str = "claude-sonnet-5") -> int: """使用 Anthropic API 统计 token 数""" client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

# 使用 count_tokens 端点
result = client.messages.count_tokens(
    model=model,
    messages=[{"role": "user", "content": text}]
)

return result.input_tokens

def count_tokens_with_system(messages: list, system: str = None, model: str = "claude-sonnet-5") -> dict: """统计系统提示词和消息中的 token 数""" client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

params = {
    "model": model,
    "messages": messages
}
if system:
    params["system"] = system

result = client.messages.count_tokens(**params)

return {
    "input_tokens": result.input_tokens,
    "system_tokens": getattr(result, "system_tokens", 0)
}

```

智能 Token 预算管理

```python from typing import List, Dict import anthropic

class TokenBudgetManager: """智能管理大模型调用的 token 预算"""

def __init__(self, model: str = "claude-sonnet-5"):
    self.model = model
    self.client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

    # 特定于模型的限制
    self.limits = {
        "claude-fable-5": {"context": 1000000, "output": 128000},
        "claude-opus-4-8": {"context": 1000000, "output": 128000},
        "claude-sonnet-5": {"context": 1000000, "output": 128000},
        "claude-haiku-4-5": {"context": 200000, "output": 64000},
    }
def get_budget(self, model: str = None) -> Dict[str, int]:
    """获取模型的令牌限制"""
    model = model or self.model
    return self.limits.get(model, {"context": 200000, "output": 64000})

def check_budget(
    self,
    messages: List[Dict],
    system: str = None,
    max_tokens: int = 4096,
    thinking_tokens: int = 0
) -> Dict[str, any]:
    """检查请求是否符合预算"""

    # 统计输入令牌数
    token_count = self.client.messages.count_tokens(
        model=self.model,
        messages=messages,
        system=system
    )

    input_tokens = token_count.input_tokens
    budget = self.get_budget()

    # 计算所需令牌总数
    total_required = input_tokens + max_tokens + thinking_tokens

    return {
        "fits_budget": total_required <= budget["context"],
        "input_tokens": input_tokens,
        "requested_output": max_tokens,
        "thinking_budget": thinking_tokens,
        "total_required": total_required,
        "context_limit": budget["context"],
        "remaining": budget["context"] - total_required,
        "utilization_pct": (total_required / budget["context"]) * 100
    }

def optimize_for_budget(
    self,
    messages: List[Dict],
    system: str = None,
    target_output: int = 4096,
    thinking_tokens: int = 0,
    priority_last_n: int = 3
) -> List[Dict]:
    """截断消息以符合预算，同时保留近期上下文"""

    budget = self.get_budget()
    available = budget["context"] - target_output - thinking_tokens

    # 始终保留系统提示词和最后 N 条消息
    preserved_messages = messages[-priority_last_n:]

    # 统计保留内容的令牌数
    preserved_count = self.client.messages.count_tokens(
        model=self.model,
        messages=preserved_messages,
        system=system
    ).input_tokens

    if preserved_count <= available:
        # 尝试包含更早的消息
        remaining = available - preserved_count
        earlier_messages = messages[:-priority_last_n]

        # 使用二分查找，确定可以容纳多少条更早的消息
        left, right = 0, len(earlier_messages)
        best_fit = 0

        while left <= right:
            mid = (left + right) // 2
            test_messages = earlier_messages[-mid:] + preserved_messages

            test_count = self.client.messages.count_tokens(
                model=self.model,
                messages=test_messages,
                system=system
            ).input_tokens

            if test_count <= available:
                best_fit = mid
                left = mid + 1
            else:
                right = mid - 1

        return earlier_messages[-best_fit:] + preserved_messages if best_fit > 0 else preserved_messages
    else:
        # 即使只保留的消息也超出预算，则截断这些消息
        return preserved_messages[-1:]  # 至少保留最后一条消息

def get_recommendations(self, budget_check: Dict) -> List[str]:
    """根据预算使用率获取建议"""
    recommendations = []

    util = budget_check["utilization_pct"]
    if util > 90:
        recommendations.append("严重：Token 使用率 >90%。请考虑缩短上下文或输出。")
        recommendations.append("仅在任务复杂度确有需要时启用扩展思考。")
    elif util > 75:
        recommendations.append("警告：Token 使用率 >75%。请监控上下文大小。")
        recommendations.append("请考虑总结较早的对话轮次。")
    elif util > 50:
        recommendations.append("Token 使用率适中。预算充足。")
    else:
        recommendations.append("Token 使用率较低。预算余量充足。")

    if budget_check["thinking_budget"] > budget_check["requested_output"]:
        recommendations.append("思考预算超过输出预算。请确认这是有意设置的。")

    return recommendations

使用示例

manager = TokenBudgetManager(model="claude-sonnet-5")

messages = [ {"role": "user", "content": "什么是 Python？"}, {"role": "assistant", "content": "Python 是一种高级编程语言……"}, {"role": "user", "content": "编写一个复杂的微服务架构"} ]

system = "你是一名资深软件架构师。"

检查预算

budget_check = manager.check_budget( messages=messages, system=system, max_tokens=8192, thinking_tokens=10000 )

print(f"是否符合预算：{budget_check['fits_budget']}") print(f"使用率：{budget_check['utilization_pct']:.1f}%") print(f"剩余 Token：{budget_check['remaining']}")

获取建议

for rec in manager.get_recommendations(budget_check): print(f"- {rec}")

按需优化

if not budget_check["fits_budget"]: optimized_messages = manager.optimize_for_budget( messages=messages, system=system, target_output=8192, thinking_tokens=10000 ) print(f"消息数从 {len(messages)} 减少到 {len(optimized_messages)}") ```

OpenAI Token 计数

```python import tiktoken

def count_tokens_openai(text: str, model: str = "gpt-4") -> int: """计算 OpenAI 模型的 Token 数量""" encoding = tiktoken.encoding_for_model(model) return len(encoding.encode(text))

def truncate_to_token_limit(text: str, max_tokens: int, model: str = "gpt-4") -> str: """截断文本，使其符合 Token 限制""" encoding = tiktoken.encoding_for_model(model) tokens = encoding.encode(text) if len(tokens) <= max_tokens: return text return encoding.decode(tokens[:max_tokens])

def split_by_tokens(text: str, chunk_size: int, model: str = "gpt-4") -> List[str]: """将文本拆分为指定 Token 大小的块""" encoding = tiktoken.encoding_for_model(model) tokens = encoding.encode(text)

chunks = []
for i in range(0, len(tokens), chunk_size):
    chunk_tokens = tokens[i:i + chunk_size]
    chunks.append(encoding.decode(chunk_tokens))

return chunks

```

交叉引用和相关技能

扩展思考

适用于需要深入分析的复杂推理任务：

• 技能： .claude/skills/extended-thinking/SKILL.md
• 适用场景： 多步推理、架构设计、数学证明

复杂推理

适用于高级问题求解和分析：

• 技能： .claude/skills/complex-reasoning/SKILL.md
• 适用场景： 系统设计、优化问题、战略规划

深度分析

适用于全面分析代码库和系统：

• 技能： .claude/skills/deep-analysis/SKILL.md
• 适用场景： 代码审查、安全审计、性能分析

编排集成

在 Golden Armada 编排系统中使用 LLM 集成时：

```python from orchestration.agent_activity_logger import AgentActivityLogger

记录 LLM API 调用以进行跟踪

logger = AgentActivityLogger()

logger.log_activity( agent_name="code-reviewer", activity_type="llm_api_call", details={ "provider": "anthropic", "model": "claude-sonnet-5", "input_tokens": 1500, "output_tokens": 800, "thinking_tokens": 3000, "extended_thinking": True, "task": "security_audit" } ) ```

模型选择指南

Claude 模型

模型	最适用于	上下文	输出	费用（每 MTok 输入/输出）
Fable 5 (claude-fable-5)	最复杂的长周期推理和 Agent 工作	1M	128K	$10 / $50
Opus 4.8 (claude-opus-4-8)	复杂推理、架构设计	1M	128K	$5 / $25
Sonnet 5 (claude-sonnet-5)	通用开发、均衡性能	1M	128K	$3 / $15
Haiku 4.5 (claude-haiku-4-5)	快速任务、简单查询、高吞吐量	200K	64K	$1 / $5

思考： 所有当前模型都支持自适应思考（{"type": "adaptive"}）；Fable 5 始终启用思考（省略 thinking 参数）。手动设置 budget_tokens 仅适用于较旧的模型。

各模型的适用场景

```python MODEL_SELECTION = { "strategic_planning": "claude-opus-4-8", "system_architecture": "claude-opus-4-8", "complex_debugging": "claude-sonnet-5", "code_review": "claude-sonnet-5", "code_generation": "claude-sonnet-5", "documentation": "claude-haiku-4-5", "simple_queries": "claude-haiku-4-5", "batch_processing": "claude-haiku-4-5" }

def select_model(task_type: str, use_thinking: bool = False) -> str: """根据任务选择合适的模型""" model = MODEL_SELECTION.get(task_type, "claude-sonnet-5")

# 请求深度推理时，优先选择更强的模型
if use_thinking and "haiku" in model:
    model = "claude-sonnet-5"

return model

```

性能优化技巧

1. 提示缓存（降低费用）

```python

缓存较大的系统提示词或上下文

response = client.messages.create( model="claude-sonnet-5", max_tokens=4096, system=[ { "type": "text", "text": large_system_prompt, "cache_control": {"type": "ephemeral"} # 缓存此内容 } ], messages=[ {"role": "user", "content": "关于该系统的问题"} ] ) ```

2. 批处理

```python async def process_batch(items: List[str], batch_size: int = 10): """分批处理项目以避免触发速率限制""" import asyncio

async def process_item(item: str):
    # 异步 LLM 调用
    return await async_llm_call(item)

results = []
for i in range(0, len(items), batch_size):
    batch = items[i:i + batch_size]
    batch_results = await asyncio.gather(*[process_item(item) for item in batch])
    results.extend(batch_results)

    # 速率限制
    if i + batch_size < len(items):
        await asyncio.sleep(1)

return results

```

3. 使用流式输出提升响应速度

```python

长时间运行的任务始终使用流式输出

with client.messages.stream( model="claude-sonnet-5", max_tokens=8192, messages=[{"role": "user", "content": complex_task}] ) as stream: for text in stream.text_stream: yield text # 将内容流式传输到 UI 或逐步处理 ```