Claude API高级用法:构建企业级AI工作流

17老师 · 2026-06-14 · 阅读约15分钟

← 返回博客列表

前言

Claude是Anthropic开发的大语言模型,以其长文本处理能力和工具调用功能著称。本文深入探索Claude API的高级用法,包括System Prompt优化、工具调用、多轮对话管理,以及如何用Claude构建企业级自动化工作流。

一、Claude API基础

1.1 安装和配置

pip install anthropic

import anthropic

client = anthropic.Anthropic(
    api_key="your-api-key"
)

# 基础对话
message = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    system="你是一个专业的助手。",
    messages=[
        {"role": "user", "content": "你好,请介绍一下你自己。"}
    ]
)
print(message.content[0].text)

1.2 流式输出

with client.messages.stream(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    system="你是一个专业的助手。",
    messages=[{"role": "user", "content": "写一首关于编程的诗"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

二、System Prompt优化

2.1 角色设定

SYSTEM_PROMPT = """
你是一个企业级AI助手,具备以下能力:

## 核心能力
1. 文档分析和总结
2. 代码审查和优化建议
3. 数据分析和可视化建议
4. 项目管理和任务分解

## 工作规范
- 使用中文回答
- 回答要结构化,使用标题和列表
- 代码示例要完整可运行
- 不确定的信息要标注
- 涉及敏感操作要提醒用户确认

## 工具使用
当你需要执行具体操作时,使用提供的工具:
- search_knowledge_base: 搜索知识库
- create_task: 创建任务
- send_notification: 发送通知

## 输出格式
- 正式回答使用Markdown格式
- 代码块使用```language标注
- 关键信息使用**加粗**
"""

2.2 Few-shot示例

messages = [
    {"role": "user", "content": "帮我分析这个代码的问题:\ndef add(a, b):\n    return a - b"},
    {"role": "assistant", "content": "## 代码分析\n\n**问题发现:**\n函数名是`add`(加法),但实际执行的是减法操作(`a - b`)。\n\n**修复建议:**\n```python\ndef add(a, b):\n    return a + b\n```\n\n**修改说明:**\n将减号`-`改为加号`+`,使函数名和功能一致。"},
    {"role": "user", "content": "帮我分析这个代码的问题:\ndef multiply(x, y):\n    return x + y"},
    {"role": "assistant", "content": "## 代码分析\n\n**问题发现:**\n函数名是`multiply`(乘法),但实际执行的是加法操作(`x + y`)。\n\n**修复建议:**\n```python\ndef multiply(x, y):\n    return x * y\n```\n\n**修改说明:**\n将加号`+`改为乘号`*`,使函数名和功能一致。"}
]

三、工具调用(Tool Use)

3.1 定义工具

import json

tools = [
    {
        "name": "search_knowledge_base",
        "description": "搜索企业知识库,查找相关文档和信息",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "搜索关键词"
                },
                "max_results": {
                    "type": "integer",
                    "description": "最大返回结果数",
                    "default": 5
                }
            },
            "required": ["query"]
        }
    },
    {
        "name": "create_task",
        "description": "在项目管理工具中创建新任务",
        "input_schema": {
            "type": "object",
            "properties": {
                "title": {
                    "type": "string",
                    "description": "任务标题"
                },
                "description": {
                    "type": "string",
                    "description": "任务描述"
                },
                "priority": {
                    "type": "string",
                    "enum": ["low", "medium", "high"],
                    "description": "优先级"
                },
                "assignee": {
                    "type": "string",
                    "description": "负责人"
                }
            },
            "required": ["title"]
        }
    },
    {
        "name": "send_notification",
        "description": "发送通知给指定用户或团队",
        "input_schema": {
            "type": "object",
            "properties": {
                "recipient": {
                    "type": "string",
                    "description": "接收者"
                },
                "message": {
                    "type": "string",
                    "description": "通知内容"
                },
                "channel": {
                    "type": "string",
                    "enum": ["email", "slack", "dingtalk"],
                    "description": "通知渠道"
                }
            },
            "required": ["recipient", "message"]
        }
    }
]

3.2 执行工具调用

def execute_tool(tool_name: str, tool_input: dict) -> str:
    """执行工具调用"""
    if tool_name == "search_knowledge_base":
        # 实现知识库搜索逻辑
        results = search_kb(tool_input["query"], tool_input.get("max_results", 5))
        return json.dumps(results, ensure_ascii=False)
    
    elif tool_name == "create_task":
        # 实现创建任务逻辑
        task = create_task_in_jira(tool_input)
        return json.dumps({"task_id": task["id"], "status": "created"})
    
    elif tool_name == "send_notification":
        # 实现发送通知逻辑
        send_msg(tool_input["recipient"], tool_input["message"], tool_input["channel"])
        return json.dumps({"status": "sent"})
    
    return json.dumps({"error": "Unknown tool"})

def chat_with_tools(user_message: str, conversation_history: list) -> str:
    """带工具调用的对话"""
    conversation_history.append({"role": "user", "content": user_message})
    
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        system=SYSTEM_PROMPT,
        tools=tools,
        messages=conversation_history
    )
    
    # 处理工具调用
    if response.stop_reason == "tool_use":
        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                result = execute_tool(block.name, block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result
                })
        
        conversation_history.append({"role": "assistant", "content": response.content})
        conversation_history.append({"role": "user", "content": tool_results})
        
        # 获取最终回复
        final_response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            system=SYSTEM_PROMPT,
            tools=tools,
            messages=conversation_history
        )
        return final_response.content[0].text
    
    return response.content[0].text

四、长文本处理

4.1 文档摘要

def summarize_document(document: str, max_length: int = 500) -> str:
    """使用Claude生成文档摘要"""
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        system="你是一个专业的文档分析助手,擅长生成简洁准确的摘要。",
        messages=[{
            "role": "user",
            "content": f"请为以下文档生成一个{max_length}字以内的摘要:\n\n{document}"
        }]
    )
    return response.content[0].text

def extract_key_info(document: str, questions: list) -> dict:
    """从文档中提取关键信息"""
    questions_text = "\n".join([f"- {q}" for q in questions])
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2048,
        system="你是一个信息提取专家,从文档中准确提取指定信息。",
        messages=[{
            "role": "user",
            "content": f"从以下文档中提取关键信息:\n\n文档:{document}\n\n需要提取的信息:\n{questions_text}\n\n请以JSON格式返回结果。"
        }]
    )
    return json.loads(response.content[0].text)

五、企业级工作流

5.1 自动化审批流程

class ApprovalWorkflow:
    """基于Claude的自动化审批工作流"""
    
    def __init__(self):
        self.client = anthropic.Anthropic()
    
    def analyze_request(self, request: dict) -> dict:
        """分析审批请求,给出建议"""
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            system="""你是一个审批分析助手。根据请求内容,分析风险并给出审批建议。
            
输出格式:
{
  "risk_level": "low|medium|high",
  "suggestion": "approve|reject|need_more_info",
  "reason": "分析原因",
  "suggested_actions": ["建议操作1", "建议操作2"]
}""",
            messages=[{
                "role": "user",
                "content": f"分析以下审批请求:\n{json.dumps(request, ensure_ascii=False)}"
            }]
        )
        return json.loads(response.content[0].text)
    
    def generate_approval_email(self, request: dict, decision: str) -> str:
        """生成审批结果邮件"""
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            system="你是一个邮件撰写助手,生成专业、清晰的审批结果通知邮件。",
            messages=[{
                "role": "user",
                "content": f"为以下审批结果生成邮件:\n请求:{json.dumps(request, ensure_ascii=False)}\n结果:{decision}"
            }]
        )
        return response.content[0].text

5.2 智能客服工作流

class SmartCustomerService:
    """基于Claude的智能客服"""
    
    def __init__(self):
        self.client = anthropic.Anthropic()
        self.conversations = {}
    
    def handle_message(self, session_id: str, user_message: str) -> str:
        """处理用户消息"""
        if session_id not in self.conversations:
            self.conversations[session_id] = []
        
        history = self.conversations[session_id]
        history.append({"role": "user", "content": user_message})
        
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            system=SYSTEM_PROMPT,
            tools=tools,
            messages=history
        )
        
        # 处理响应和工具调用
        assistant_message = response.content[0].text if response.stop_reason != "tool_use" else self._handle_tools(response, history)
        
        history.append({"role": "assistant", "content": assistant_message})
        
        # 保持对话历史在合理范围
        if len(history) > 20:
            history[:] = history[-20:]
        
        return assistant_message

六、性能优化

优化建议:
  • Prompt缓存:Claude支持Prompt缓存,重复使用相同的System Prompt可以降低成本和延迟
  • 批量处理:使用Batch API处理大量请求,成本降低50%
  • 流式输出:使用流式输出提升用户体验
  • 错误重试:实现指数退避重试机制
  • 成本控制:监控API使用量,设置预算告警

总结

Claude API提供了强大的AI能力,通过工具调用、长文本处理、自动化工作流等功能,可以构建企业级的AI应用。掌握这些高级用法,可以让AI真正赋能业务。如果需要Claude API集成服务,欢迎联系17老师。

关于17老师:AI应用·数字化管理·全栈开发·网络安全全栈专家。联系邮箱:j.d88888888@qq.com,微信:AFIST17
上一篇
Electron桌面应用开发
返回
博客列表