Temporal Python SDK多语言服务：服务工具集成

【免费下载链接】sdk-python Temporal Python SDK 项目地址: https://gitcode.com/GitHub_Trending/sd/sdk-python

1. 引言

在现代软件开发中，多语言服务集成和工具调用是构建复杂应用的关键挑战。开发者常常需要处理不同服务之间的通信、数据格式转换以及分布式追踪等问题。Temporal Python SDK提供了一套强大的解决方案，帮助开发者轻松实现多语言服务集成和工具调用，从而提高开发效率和系统可靠性。

本文将详细介绍如何使用Temporal Python SDK进行多语言服务集成，重点关注OpenAI Agents的集成方案。我们将从核心组件、集成步骤、使用示例到高级特性，全面解析Temporal Python SDK在多语言服务集成方面的强大功能。

2. 核心组件解析

Temporal Python SDK的多语言服务集成主要依赖于以下核心组件：

2.1 OpenAIAgentsPlugin

OpenAIAgentsPlugin是Temporal Python SDK提供的一个插件，用于无缝集成OpenAI Agents SDK与Temporal工作流。该插件自动配置必要的拦截器、活动和数据转换器，使OpenAI Agents能够在Temporal工作流中运行，并提供适当的追踪和模型执行能力。

该插件的主要功能包括：

• 配置Pydantic数据转换器，实现类型安全的序列化
• 设置OpenAI agent交互的追踪拦截器
• 注册模型执行活动
• 自动注册MCP服务器活动并管理其生命周期
• 在工作器执行期间管理OpenAI agent运行时覆盖

2.2 TemporalOpenAIRunner

TemporalOpenAIRunner是一个自定义的agent运行器，它将OpenAI agent的执行委托给Temporal活动。这使得agent的执行可以在Temporal的分布式环境中进行，享受Temporal提供的可靠性和可观察性。

2.3 TemporalTraceProvider

TemporalTraceProvider是一个自定义的追踪提供器，它将OpenAI agent的追踪信息与Temporal的工作流追踪集成。这使得开发者可以在Temporal的UI中查看agent的执行轨迹，从而更好地调试和优化agent的行为。

2.4 ModelActivity

ModelActivity是一个Temporal活动，用于执行OpenAI模型调用。它处理模型参数的验证、模型的调用以及结果的返回，确保模型调用的可靠性和可追踪性。

3. 集成步骤

使用Temporal Python SDK集成OpenAI Agents主要包括以下步骤：

3.1 安装依赖

首先，需要安装Temporal Python SDK和OpenAI Agents SDK：

pip install temporalio openai-agents

3.2 配置ModelActivityParameters

ModelActivityParameters用于配置模型活动的参数，如超时时间、重试策略等。例如：

from temporalio.contrib.openai_agents import ModelActivityParameters
from temporalio.common import RetryPolicy
from datetime import timedelta

model_params = ModelActivityParameters(
    start_to_close_timeout=timedelta(seconds=30),
    retry_policy=RetryPolicy(maximum_attempts=3)
)

3.3 创建OpenAIAgentsPlugin

使用配置好的ModelActivityParameters创建OpenAIAgentsPlugin实例：

from temporalio.contrib.openai_agents import OpenAIAgentsPlugin

plugin = OpenAIAgentsPlugin(model_params=model_params)

3.4 配置Temporal客户端和工作器

将插件添加到Temporal客户端和工作器的配置中：

import temporalio.client
import temporalio.worker

# 连接到Temporal服务
client = await temporalio.client.Client.connect(
    "localhost:7233",
    plugins=[plugin]
)

# 创建工作器
worker = temporalio.worker.Worker(
    client,
    task_queue="openai-agent-task-queue",
    workflows=[MyAgentWorkflow],
    plugins=[plugin]
)

# 启动工作器
await worker.run()

3.5 实现agent工作流

创建一个使用OpenAI agent的Temporal工作流：

from temporalio import workflow
from agents import Agent, AgentSettings, ModelSettings

@workflow.defn
class MyAgentWorkflow:
    @workflow.run
    async def run(self, input: str) -> str:
        agent = Agent(
            settings=AgentSettings(
                model_settings=ModelSettings(model_name="gpt-4")
            )
        )
        
        result = await agent.run(input=input)
        return result.content

4. 使用示例

以下是一个完整的示例，展示如何使用Temporal Python SDK集成OpenAI Agents：

4.1 配置插件和MCP服务器

from temporalio.contrib.openai_agents import OpenAIAgentsPlugin, ModelActivityParameters, StatelessMCPServerProvider
from agents.mcp import MCPServerStdio
from datetime import timedelta

# 配置模型参数
model_params = ModelActivityParameters(
    start_to_close_timeout=timedelta(seconds=30),
    retry_policy=RetryPolicy(maximum_attempts=3)
)

# 创建MCP服务器
filesystem_server = StatelessMCPServerProvider(MCPServerStdio(
    name="Filesystem Server",
    params={"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "."]}
))

# 创建插件
plugin = OpenAIAgentsPlugin(
    model_params=model_params,
    mcp_server_providers=[filesystem_server]
)

4.2 实现一个使用工具的agent工作流

from temporalio import workflow
from agents import Agent, AgentSettings, ModelSettings, Tool
from pydantic import BaseModel

class CalculatorInput(BaseModel):
    expression: str

class CalculatorOutput(BaseModel):
    result: float

async def calculator_tool(input: CalculatorInput) -> CalculatorOutput:
    """一个简单的计算器工具，用于计算数学表达式"""
    return CalculatorOutput(result=eval(input.expression))

@workflow.defn
class CalculatorAgentWorkflow:
    @workflow.run
    async def run(self, input: str) -> str:
        # 定义工具
        tools = [
            Tool(
                name="calculator",
                description="用于计算数学表达式",
                input_schema=CalculatorInput,
                output_schema=CalculatorOutput,
                func=calculator_tool
            )
        ]
        
        # 创建agent
        agent = Agent(
            settings=AgentSettings(
                model_settings=ModelSettings(model_name="gpt-4"),
                tools=tools
            )
        )
        
        # 运行agent
        result = await agent.run(input=input)
        return result.content

4.3 启动工作器并运行工作流

import temporalio.client
import temporalio.worker

# 连接到Temporal服务
client = await temporalio.client.Client.connect(
    "localhost:7233",
    plugins=[plugin]
)

# 创建工作器
worker = temporalio.worker.Worker(
    client,
    task_queue="calculator-agent-task-queue",
    workflows=[CalculatorAgentWorkflow],
    plugins=[plugin]
)

# 启动工作器（在单独的线程中运行）
import threading
threading.Thread(target=lambda: asyncio.run(worker.run()), daemon=True).start()

# 运行工作流
result = await client.execute_workflow(
    CalculatorAgentWorkflow.run,
    "计算 2 + 2 * 3",
    id="calculator-agent-workflow",
    task_queue="calculator-agent-task-queue"
)

print(f"工作流结果: {result}")  # 输出: "工作流结果: 8"

5. 高级特性

5.1 自定义模型提供器

Temporal Python SDK允许开发者使用自定义的模型提供器，这对于测试或自定义模型集成非常有用。例如，我们可以创建一个测试模型提供器，用于在不调用实际OpenAI API的情况下测试工作流：

from temporalio.contrib.openai_agents import TestModelProvider, TestModel
from agents import ModelResponse, ModelSettings

class MyTestModel(TestModel):
    async def get_response(self, *args, **kwargs) -> ModelResponse:
        return ModelResponse(content="这是一个测试响应")

plugin = OpenAIAgentsPlugin(
    model_params=model_params,
    model_provider=TestModelProvider(MyTestModel())
)

5.2 分布式追踪

Temporal Python SDK与OpenAI Agents的集成提供了完整的分布式追踪能力。通过TemporalTraceProvider和OpenAIAgentsTracingInterceptor，开发者可以在Temporal的UI中查看agent的执行轨迹，包括工具调用、模型调用等。

5.3 MCP服务器集成

Temporal Python SDK支持Model Context Protocol (MCP)服务器的集成，这使得agent可以与外部系统进行交互。通过StatelessMCPServerProvider和StatefulMCPServerProvider，开发者可以轻松地将MCP服务器集成到Temporal工作流中。

5.4 工作流重放

Temporal Python SDK提供了工作流重放功能，可以重放agent的执行过程，这对于调试和优化agent的行为非常有用。通过Replayer类，开发者可以重放工作流历史，重现agent的执行过程。

from temporalio.worker import Replayer

# 创建重放器
replayer = Replayer(
    workflows=[CalculatorAgentWorkflow],
    plugins=[plugin]
)

# 重放工作流历史
history = await client.get_workflow_history("calculator-agent-workflow")
await replayer.replay_workflow(history)

6. 最佳实践

6.1 设置适当的超时和重试策略

当集成外部服务时，设置适当的超时和重试策略非常重要。通过ModelActivityParameters，开发者可以为模型调用设置超时和重试策略，以应对网络问题或服务不可用的情况。

6.2 使用工作流版本控制

当更新agent的行为时，使用Temporal的工作流版本控制功能可以确保旧版本的工作流继续正常运行，同时新版本的工作流使用更新后的agent。

6.3 监控和日志

Temporal Python SDK提供了丰富的监控和日志功能。开发者应该监控agent的执行指标，如模型调用的延迟、成功率等，并记录详细的日志，以便在出现问题时进行调试。

6.4 测试策略

对于集成了OpenAI Agents的Temporal工作流，建议采用以下测试策略：

1. 使用TestModelProvider进行单元测试
2. 使用Temporal的测试框架进行集成测试
3. 使用工作流重放功能进行回归测试

7. 总结

Temporal Python SDK为多语言服务集成和工具调用提供了强大的支持。通过OpenAIAgentsPlugin等核心组件，开发者可以轻松地将OpenAI Agents集成到Temporal工作流中，享受Temporal提供的可靠性、可观察性和可扩展性。

无论是构建复杂的AI助手，还是集成各种外部服务，Temporal Python SDK都能帮助开发者简化开发流程，提高系统的可靠性和可维护性。

希望本文能够帮助开发者更好地理解和使用Temporal Python SDK进行多语言服务集成。如有任何问题或建议，欢迎在项目的GitHub仓库中提出。

【免费下载链接】sdk-python Temporal Python SDK 项目地址: https://gitcode.com/GitHub_Trending/sd/sdk-python