Python 调用 OpenAI / Claude / Gemini API 实战：从批量处理到 AI Agent 与 LangChain 搭建

1
import os
2
import httpx
3
from dotenv import load_dotenv
4
from openai import OpenAI
5
from anthropic import Anthropic
6

7
# 加载本地 .env 环境变量
8
load_dotenv()
9

10
proxy_endpoint = os.getenv("LOCAL_PROXY_URL", "http://127.0.0.1:7890")
11

12
# 1. 为底层传输构造一个高韧性 HTTP 连接客户端
13
custom_http_client = httpx.Client(
14
    proxy=proxy_endpoint,
15
    timeout=httpx.Timeout(connect=10.0, read=60.0, write=10.0, pool=10.0),
16
    limits=httpx.Limits(max_keepalive_connections=50, max_connections=100)
17
)
18

19
# 2. 注入 OpenAI 客户端实例
20
openai_client = OpenAI(
21
    api_key=os.getenv("OPENAI_API_KEY"),
22
    http_client=custom_http_client
23
)
24

25
# 3. 注入 Anthropic 客户端实例
26
anthropic_client = Anthropic(
27
    api_key=os.getenv("ANTHROPIC_API_KEY"),
28
    http_client=custom_http_client
29
)
30

31
print("[✓] AI 客户端底层通信通道与凭据初始化就绪！")

1
import os
2
import httpx
3
from dotenv import load_dotenv
4

5
load_dotenv()
6

7
def verify_ai_provider_connectivity(proxy_url=None):
8
    """
9
    自检各主流模型端点的 TCP 握手与基础服务连通性
10
    """
11
    providers = {
12
        "OpenAI": "https://api.openai.com/v1/models",
13
        "Anthropic": "https://api.anthropic.com/v1/messages",
14
        "Google Gemini": "https://generativelanguage.googleapis.com"
15
    }
16

17
    client = httpx.Client(proxy=proxy_url, timeout=5.0)
18
    diagnostic_report = {}
19

20
    for name, url in providers.items():
21
        try:
22
            # 发起轻量 OPTIONS 或 HEAD 探测握手连通性
23
            resp = client.head(url)
24
            # 只要能够返回 HTTP 状态码（即便 401 未认证），均证明底层传输与 TLS 握手彻底畅通
25
            diagnostic_report[name] = {"reachable": True, "http_code": resp.status_code}
26
            print(f"[✓] {name} 节点网络握手成功，HTTP 状态响应码: {resp.status_code}")
27
        except httpx.ConnectError:
28
            diagnostic_report[name] = {"reachable": False, "reason": "TCP连接拒绝或握手超时"}
29
            print(f"[×] {name} 连接失败: 物理路由不可达或被阻断，请检查代理通道！")
30
        except Exception as err:
31
            diagnostic_report[name] = {"reachable": False, "reason": str(err)}
32
            print(f"[×] {name} 探测异常: {err}")
33

34
    client.close()
35
    return diagnostic_report
36

37
if __name__ == "__main__":
38
    proxy = os.getenv("LOCAL_PROXY_URL")
39
    verify_ai_provider_connectivity(proxy)

1
import asyncio
2
import os
3
from dotenv import load_dotenv
4
from openai import AsyncOpenAI
5
import httpx
6

7
load_dotenv()
8

9
async def process_single_task(async_client, semaphore, task_id, text_payload):
10
    # 利用信号量（Semaphore）严格压制瞬时并发上限，避免触碰平台的 429 速率限制
11
    async with semaphore:
12
        try:
13
            response = await async_client.chat.completions.create(
14
                model="gpt-4o-mini",
15
                messages=[
16
                    {"role": "system", "content": "你是一个高效的信息提取助手，仅输出核心总结，字数控制在30字以内。"},
17
                    {"role": "user", "content": text_payload}
18
                ],
19
                temperature=0.3,
20
                max_tokens=100
21
            )
22
            result = response.choices[0].message.content.strip()
23
            return {"id": task_id, "status": "SUCCESS", "result": result}
24
        except Exception as e:
25
            return {"id": task_id, "status": "FAILED", "error": str(e)}
26

27
async def run_batch_processing_pipeline(tasks_data, max_concurrency=15):
28
    proxy_url = os.getenv("LOCAL_PROXY_URL")
29
    http_client = httpx.AsyncClient(proxy=proxy_url) if proxy_url else None
30

31
    async_client = AsyncOpenAI(
32
        api_key=os.getenv("OPENAI_API_KEY"),
33
        http_client=http_client
34
    )
35

36
    # 限制同时最大在途请求数为 15
37
    sem = asyncio.Semaphore(max_concurrency)
38

39
    print(f"[*] 启动异步并发流水线，任务队列总量: {len(tasks_data)}，并发槽位: {max_concurrency}")
40

41
    tasks = [
42
        process_single_task(async_client, sem, item["id"], item["content"])
43
        for item in tasks_data
44
    ]
45

46
    # 并发等待全量任务完成
47
    results = await asyncio.gather(*tasks)
48
    if http_client:
49
        await http_client.aclose()
50
    return results
51

52
# 示范执行入口
53
if __name__ == "__main__":
54
    mock_dataset = [
55
        {"id": i, "content": f"2026年第{i}季度业务报告：服务器集群负载整体下降15%，系统高可用达到四个九。"}
56
        for i in range(1, 21)
57
    ]
58
    output_records = asyncio.run(run_batch_processing_pipeline(mock_dataset, max_concurrency=5))
59
    print(f"[✓] 批量处理完毕，首条样本产出: {output_records[0]}")

1
import os
2
import json
3
import time
4
from openai import OpenAI
5
from dotenv import load_dotenv
6

7
load_dotenv()
8
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
9

10
def execute_official_batch_pipeline():
11
    batch_input_filename = "offline_tasks_2026.jsonl"
12

13
    # 第一步：构建标准 JSONL 任务文件（每行包含一个独立的自定义请求结构）
14
    print("[1/4] 正在构造离线批处理输入 JSONL 资产...")
15
    tasks_pool = [
16
        {
17
            "custom_id": f"task-req-{i}",
18
            "method": "POST",
19
            "url": "/v1/chat/completions",
20
            "body": {
21
                "model": "gpt-4o-mini",
22
                "messages": [
23
                    {"role": "system", "content": "提取文本的核心关键词，以逗号分隔。"},
24
                    {"role": "user", "content": f"样本内容段落 {i}：持续集成系统在构建阶段自动执行静态安全扫描与代码审计。"}
25
                ],
26
                "max_tokens": 50
27
            }
28
        }
29
        for i in range(1, 6)
30
    ]
31

32
    with open(batch_input_filename, "w", encoding="utf-8") as f:
33
        for item in tasks_pool:
34
            f.write(json.dumps(item, ensure_ascii=False) + "\n")
35

36
    # 第二步：将本地文件上传到 OpenAI 专用的 Batch 文件存储区
37
    print("[2/4] 上传 JSONL 任务资产到 OpenAI 云端存储...")
38
    with open(batch_input_filename, "rb") as f:
39
        uploaded_file = client.files.create(file=f, purpose="batch")
40
    print(f" -> 文件上传成功，获得云端资产 ID: {uploaded_file.id}")
41

42
    # 第三步：基于上传的文件创建异步批处理作业
43
    print("[3/4] 提交并创建 Batch 批处理作业流水线...")
44
    batch_job = client.batches.create(
45
        input_file_id=uploaded_file.id,
46
        endpoint="/v1/chat/completions",
47
        completion_window="24h", # 官方保证在 24 小时内处理完毕，享受 50% 折扣
48
        metadata={"project": "Enterprise-Archive-2026"}
49
    )
50
    print(f" -> 作业成功提交，Batch Job ID: {batch_job.id} | 初始状态: {batch_job.status}")
51

52
    # 第四步：轮询状态并提取输出结果
53
    print("[4/4] 正在轮询批处理进度...")
54
    while True:
55
        status_check = client.batches.retrieve(batch_job.id)
56
        print(f" -> 进度心跳: 状态=[{status_check.status}] | 已处理: {status_check.request_counts.completed}/{status_check.request_counts.total}")
57

58
        if status_check.status == "completed":
59
            print("[✓] 离线批处理全量完成！正在拉取最终结果输出文件...")
60
            output_file_id = status_check.output_file_id
61
            content_response = client.files.content(output_file_id)
62

63
            with open("batch_results_output.jsonl", "wb") as out_f:
64
                out_f.write(content_response.content)
65
            print("[✓] 批处理结果已成功持久化至 batch_results_output.jsonl")
66
            break
67
        elif status_check.status in ["failed", "cancelled", "expired"]:
68
            print(f"[×] 批处理作业未能平稳完成，终止状态: {status_check.status}")
69
            break
70

71
        time.sleep(10)
72

73
if __name__ == "__main__":
74
    # 执行批处理流程演示
75
    execute_official_batch_pipeline()

1
import asyncio
2
import time
3
import os
4
from dotenv import load_dotenv
5
from openai import AsyncOpenAI
6

7
load_dotenv()
8

9
async def stream_chat_completion(prompt_text):
10
    client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))
11

12
    print("[*] 正在与云端大模型握手建立流式通道...")
13
    start_time = time.time()
14
    first_token_received = False
15
    full_response_text = []
16

17
    # 启用 stream=True 参数激活 SSE 流式推送
18
    stream_generator = await client.chat.completions.create(
19
        model="gpt-4o",
20
        messages=[
21
            {"role": "user", "content": prompt_text}
22
        ],
23
        stream=True,
24
        temperature=0.7
25
    )
26

27
    async for chunk in stream_generator:
28
        # 首字节抵达时间监控
29
        if not first_token_received:
30
            ttft = round((time.time() - start_time) * 1000, 2)
31
            print(f"\n[✓] 流式数据通道建立成功！首字延迟(TTFT): {ttft} ms\n--- 正文开始 ---")
32
            first_token_received = True
33

34
        # 从增量数据块中提取内容
35
        delta_content = chunk.choices[0].delta.content
36
        if delta_content:
37
            print(delta_content, end="", flush=True)
38
            full_response_text.append(delta_content)
39

40
    print("\n--- 正文结束 ---")
41
    total_time = round(time.time() - start_time, 2)
42
    print(f"[*] 全流程耗时: {total_time} 秒，生成完整文本字数: {len(''.join(full_response_text))}")
43

44
if __name__ == "__main__":
45
    test_query = "请详细分析企业在混合云架构下实施自动化运维的三个核心优势。"
46
    asyncio.run(stream_chat_completion(test_query))

1
import os
2
from typing import List, Optional
3
from pydantic import BaseModel, Field
4
from openai import OpenAI
5
from dotenv import load_dotenv
6

7
load_dotenv()
8

9
# 1. 使用 Pydantic 显式定义下游系统严苛需要的强类型实体结构
10
class VulnerabilityReport(BaseModel):
11
    cve_id: str = Field(description="标准化通用漏洞编号，例如 CVE-2026-1024")
12
    severity_level: str = Field(description="漏洞危险等级，仅限 CRITICAL / HIGH / MEDIUM / LOW")
13
    affected_components: List[str] = Field(description="受波及的核心组件或类库列表")
14
    cvss_score: float = Field(description="CVSS 基础评分，范围在 0.0 到 10.0 之间")
15
    mitigation_steps: str = Field(description="应急处置与缓解建议措施")
16

17
class SystemAuditPayload(BaseModel):
18
    scan_timestamp: str = Field(description="扫描分析完成的标准时间戳")
19
    total_risks_found: int = Field(description="发现的风险总量")
20
    findings: List[VulnerabilityReport] = Field(description="检测出的结构化漏洞清单")
21

22
# 2. 实例化客户端并调用 parse 接口
23
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
24

25
unstructured_log = """
26
2026-03-08 14:30:15 安全告警：检测到网关底层存在缓冲区溢出高危风险，编号为 CVE-2026-9812。
27
该问题直接波及 openssl 和 libcurl 模块，CVSS 综合评估评分为 9.8 分，属于紧急严重级别。
28
建议立即升级系统补丁至 3.2.1 版本，并在防火墙层临时拦截外部恶意探测包。
29
"""
30

31
print("[*] 正在向模型投递非结构化日志并强制执行 Pydantic 架构约束...")
32

33
# 采用 beta.chat.completions.parse 直接获得强类型对象
34
completion = client.beta.chat.completions.parse(
35
    model="gpt-4o-2024-08-06",
36
    messages=[
37
        {"role": "system", "content": "你是一个资深网络安全分析审计专家，严格提取日志中的安全要素。"},
38
        {"role": "user", "content": unstructured_log}
39
    ],
40
    response_format=SystemAuditPayload,
41
)
42

43
parsed_data: SystemAuditPayload = completion.choices[0].message.parsed
44

45
# 此时 parsed_data 是纯正的 Python Pydantic 强类型对象，直接享受 IDE 自动补全
46
print(f"[✓] 解析成功！扫描时间: {parsed_data.scan_timestamp} | 风险总量: {parsed_data.total_risks_found}")
47
for vuln in parsed_data.findings:
48
    print(f" -> 漏洞: {vuln.cve_id} | 等级: {vuln.severity_level} | 评分: {vuln.cvss_score}")

1
import re
2
import json
3

4
def extract_and_heal_json_payload(raw_model_output: str) -> dict:
5
    """
6
    清洗大模型输出中包裹的 Markdown 语法块，并容错修补轻微的语法损伤
7
    """
8
    # 1. 正则剥离 Markdown 代码块标记（```json ... ```）
9
    json_block_match = re.search(r'```(?:json)?\s*([\s\S]*?)\s*```', raw_model_output)
10
    candidate_text = json_block_match.group(1) if json_block_match else raw_model_output.strip()
11

12
    # 2. 尝试标准解析
13
    try:
14
        return json.loads(candidate_text)
15
    except json.JSONDecodeError:
16
        pass
17

18
    # 3. 常见小瑕疵自动修补：处理尾部多余逗号与未闭合的大括号
19
    cleaned = re.sub(r',\s*([\]}])', r'\1', candidate_text)
20
    if not cleaned.endswith('}') and cleaned.count('{') > cleaned.count('}'):
21
        cleaned += '}'
22

23
    try:
24
        return json.loads(cleaned)
25
    except json.JSONDecodeError as err:
26
        raise ValueError(f"大模型生成的 JSON 存在严重结构性破坏，无法完成自动自愈: {err}\n原始文本: {candidate_text}")

1
import os
2
import json
3
import psutil
4
from openai import OpenAI
5
from dotenv import load_dotenv
6

7
load_dotenv()
8
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
9

10
# 1. 编写本地真实的物理执行函数
11
def query_system_hardware_metrics(metric_type: str) -> str:
12
    """获取当前宿主机的真实硬件负载监控数据"""
13
    if metric_type == "cpu":
14
        usage = psutil.cpu_percent(interval=1)
15
        return json.dumps({"metric": "cpu_percent", "value": f"{usage}%"})
16
    elif metric_type == "memory":
17
        mem = psutil.virtual_memory()
18
        return json.dumps({
19
            "metric": "virtual_memory",
20
            "total_gb": round(mem.total / (1024**3), 2),
21
            "used_percent": f"{mem.percent}%"
22
        })
23
    else:
24
        return json.dumps({"error": f"不支持的监控指标项: {metric_type}"})
25

26
# 2. 构造面向模型的标准工具描述定义
27
tools_definition = [
28
    {
29
        "type": "function",
30
        "function": {
31
            "name": "query_system_hardware_metrics",
32
            "description": "查询当前本地操作系统的 CPU 实时利用率或物理内存消耗指标",
33
            "parameters": {
34
                "type": "object",
35
                "properties": {
36
                    "metric_type": {
37
                        "type": "string",
38
                        "enum": ["cpu", "memory"],
39
                        "description": "需要查询的硬件指标类型"
40
                    }
41
                },
42
                "required": ["metric_type"]
43
            }
44
        }
45
    }
46
]
47

48
# 3. 发起初次多轮对话
49
dialog_history = [
50
    {"role": "user", "content": "请检查一下我们服务器目前的物理内存使用情况，如果超过80%请给出警告。"}
51
]
52

53
print("[*] 正在将用户意图与可用工具清单投递给模型...")
54
first_response = client.chat.completions.create(
55
    model="gpt-4o",
56
    messages=dialog_history,
57
    tools=tools_definition,
58
    tool_choice="auto"
59
)
60

61
response_message = first_response.choices[0].message
62
tool_calls = response_message.tool_calls
63

64
# 4. 判断模型是否主动决定调用本地工具
65
if tool_calls:
66
    print(f"[✓] 模型决策触发工具调用！调用目标: {tool_calls[0].function.name}")
67
    # 将模型的半成品回复压入对话上下文
68
    dialog_history.append(response_message)
69

70
    for tool_call in tool_calls:
71
        function_name = tool_call.function.name
72
        arguments = json.loads(tool_call.function.arguments)
73

74
        # 路由到本地真实执行逻辑
75
        if function_name == "query_system_hardware_metrics":
76
            execution_result = query_system_hardware_metrics(
77
                metric_type=arguments.get("metric_type")
78
            )
79
            print(f" -> 本地函数实际执行完毕，捕获物理数据: {execution_result}")
80

81
            # 将执行产物作为角色为 tool 的消息追加至上下文
82
            dialog_history.append({
83
                "role": "tool",
84
                "tool_call_id": tool_call.id,
85
                "content": execution_result
86
            })
87

88
    # 5. 将包含真实执行数据的完整上下文再次提交给模型，生成最终总结
89
    print("[*] 正在将工具执行产物喂回模型生成最终自然语言报告...")
90
    second_response = client.chat.completions.create(
91
        model="gpt-4o",
92
        messages=dialog_history
93
    )
94
    final_answer = second_response.choices[0].message.content
95
    print(f"\n[最终回答] {final_answer}")
96
else:
97
    print("[*] 模型认为无需调用工具，直接给出了答复:", response_message.content)

1
from langchain_core.prompts import ChatPromptTemplate
2
from langchain_core.output_parsers import StrOutputParser
3
from langchain_openai import ChatOpenAI
4

5
# 声明模型基础设施（统一抽象）
6
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
7

8
prompt = ChatPromptTemplate.from_messages([
9
    ("system", "你是一名顶级代码重构专家，用一句话指出代码的架构缺陷。"),
10
    ("user", "{code_snippet}")
11
])
12

13
# 纯正的 LCEL 链式结构：输入 -> 提示词装配 -> 模型推理 -> 纯文本解析
14
refactor_chain = prompt | llm | StrOutputParser()
15

16
# 优雅调用执行
17
advice = refactor_chain.invoke({"code_snippet": "def f(x): global a; a = x + 10; return a"})
18
print(f"[*] 架构审计意见: {advice}")

1
from typing import TypedDict, Annotated, Sequence
2
import operator
3
import os
4
from dotenv import load_dotenv
5
from langchain_core.messages import BaseMessage, HumanMessage, ToolMessage
6
from langchain_core.tools import tool
7
from langchain_openai import ChatOpenAI
8
from langgraph.graph import StateGraph, END
9
from langgraph.checkpoint.memory import MemorySaver
10

11
load_dotenv()
12

13
# 1. 定义智能体具备的真实生产工具
14
@tool
15
def calculate_system_capacity(nodes_count: int, pod_per_node: int) -> int:
16
    """计算集群在当前物理节点总数和单机容器 Pod 密度下的总承载量上限"""
17
    return nodes_count * pod_per_node
18

19
@tool
20
def query_datacenter_power_status(cluster_name: str) -> str:
21
    """查询指定机房集群当前的供电与机柜能耗冗余指标"""
22
    # 模拟真实机房指标库交互
23
    return f"集群 [{cluster_name}] 当前供电负荷 62%，机房 PUE 指标 1.18，处于安全裕度区间。"
24

25
tools = [calculate_system_capacity, query_datacenter_power_status]
26
tool_map = {t.name: t for t in tools}
27

28
# 2. 定义 Agent 的共享全局状态（State）
29
class AgentState(TypedDict):
30
    # 使用 operator.add 声明列表为累加追加模式，保证多轮对话历史不被覆写丢失
31
    messages: Annotated[Sequence[BaseMessage], operator.add]
32

33
# 3. 构造模型与图节点
34
model = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools)
35

36
def call_model_node(state: AgentState):
37
    """思考节点：模型阅读全量消息历史并决定是给出结论还是指派工具"""
38
    response = model.invoke(state["messages"])
39
    return {"messages": [response]}
40

41
def execute_tools_node(state: AgentState):
42
    """行动节点：在本地受控沙箱内安全调用模型指派的工具"""
43
    last_message = state["messages"][-1]
44
    tool_messages = []
45
    for tool_call in last_message.tool_calls:
46
        tool_name = tool_call["name"]
47
        tool_args = tool_call["args"]
48
        if tool_name in tool_map:
49
            selected_tool = tool_map[tool_name]
50
            output = selected_tool.invoke(tool_args)
51
            tool_messages.append(ToolMessage(content=str(output), tool_call_id=tool_call["id"]))
52
    return {"messages": tool_messages}
53

54
def should_continue_router(state: AgentState):
55
    """条件路由分支：判断是否需要继续执行工具，或是结束当前循环"""
56
    last_message = state["messages"][-1]
57
    if last_message.tool_calls:
58
        return "tools_actor"
59
    return END
60

61
# 4. 组装 LangGraph 状态机并挂载持久化检查点
62
workflow = StateGraph(AgentState)
63
workflow.add_node("agent_brain", call_model_node)
64
workflow.add_node("tools_actor", execute_tools_node)
65

66
workflow.set_entry_point("agent_brain")
67
workflow.add_conditional_edges("agent_brain", should_continue_router, {
68
    "tools_actor": "tools_actor",
69
    END: END
70
})
71
workflow.add_edge("tools_actor", "agent_brain") # 工具执行完自动环回大脑形成反思闭环
72

73
# 注入内存状态持久化检查点（支持多会话记忆隔离）
74
memory_checkpointer = MemorySaver()
75
react_agent = workflow.compile(checkpointer=memory_checkpointer)
76

77
# 5. 触发具备复杂推演与上下文关联的多轮会话任务
78
if __name__ == "__main__":
79
    print("[*] 正在激活基于 LangGraph 的自主演算智能体...")
80
    # 模拟线程会话 ID（Thread ID），实现用户级状态隔离
81
    session_config = {"configurable": {"thread_id": "session-prod-9812"}}
82

83
    first_query = "我们目前 A 机房有 64 台服务器，每台承载 40 个 Pod。请帮我计算总容量，并评估 A 机房能耗状态。"
84
    events = react_agent.invoke(
85
        {"messages": [HumanMessage(content=first_query)]},
86
        config=session_config
87
    )
88
    print(f"\n[Agent 第一轮推演]\n{events['messages'][-1].content}")
89

90
    # 第二轮自然语言追问（依靠检查点自动继承上一轮的计算上下文）
91
    second_query = "如果我们将服务器扩容到 100 台，总容量会变成多少？"
92
    follow_up_events = react_agent.invoke(
93
        {"messages": [HumanMessage(content=second_query)]},
94
        config=session_config
95
    )
96
    print(f"\n[Agent 继承记忆后的第二轮追问回答]\n{follow_up_events['messages'][-1].content}")

1
# 必须使用 asyncio.to_thread 卸载同步阻塞操作，防止阻塞主事件循环
2
import asyncio
3
from openai import OpenAI
4

5
sync_client = OpenAI()
6

7
async def safe_sync_wrapper(prompt: str) -> str:
8
    """在独立后台工作线程中安全执行同步调用，释放主事件循环"""
9
    loop = asyncio.get_running_loop()
10
    response = await loop.run_in_executor(
11
        None,
12
        lambda: sync_client.chat.completions.create(
13
            model="gpt-4o-mini",
14
            messages=[{"role": "user", "content": prompt}]
15
        )
16
    )
17
    return response.choices[0].message.content or ""