title: “Conversation API（Python）- 推荐用法” linkTitle: “对话” weight: 11000 type: docs description: 推荐在 Python 中配合或不配合工具使用 Dapr Conversation API 的模式，包括多轮流程与安全指引。

Dapr Conversation API 当前仍处于 alpha 阶段。本文给出在 Python SDK 中高效使用它的推荐最小模式：

前提条件

如需完整的端到端流程与提供程序配置，参见：

基础对话（无工具）

from dapr.clients import DaprClient
from dapr.clients.grpc import conversation

# 构建单轮 Alpha2 输入
user_msg = conversation.create_user_message("什么是 Dapr？")
alpha2_input = conversation.ConversationInputAlpha2(messages=[user_msg])

with DaprClient() as client:
    resp = client.converse_alpha2(
        name="echo",  # 替换为你的 LLM 组件名称
        inputs=[alpha2_input],
        temperature=1,
    )

    for msg in resp.to_assistant_messages():
        if msg.of_assistant.content:
            print(msg.of_assistant.content[0].text)

要点：

工具：基于装饰器（推荐）

基于装饰器的工具方式更整洁、也更顺手。定义函数时应写清晰的类型提示与详细 docstring；这对 LLM 理解何时以及如何调用该工具很重要； 然后使用 @conversation.tool 进行装饰。注册后的工具可以传给 LLM，并通过工具调用来执行。

from dapr.clients import DaprClient
from dapr.clients.grpc import conversation

@conversation.tool
def get_weather(location: str, unit: str = 'fahrenheit') -> str:
    """获取指定地点的当前天气。"""
    # 请替换为真实实现
    return f"Weather in {location} (unit={unit})"

user_msg = conversation.create_user_message("巴黎的天气怎么样？")
alpha2_input = conversation.ConversationInputAlpha2(messages=[user_msg])

with DaprClient() as client:
    response = client.converse_alpha2(
        name="openai",  # 你的 LLM 组件
        inputs=[alpha2_input],
        tools=conversation.get_registered_tools(),  # 由 @conversation.tool 注册的工具
        tool_choice='auto',
        temperature=1,
    )

    # 检查 assistant 消息，包括其中的工具调用
    for msg in response.to_assistant_messages():
        if msg.of_assistant.tool_calls:
            for tc in msg.of_assistant.tool_calls:
                print(f"Tool call: {tc.function.name} args={tc.function.arguments}")
        elif msg.of_assistant.content:
            print(msg.of_assistant.content[0].text)

说明：

使用工具的最小多轮流程

这是处理带工具对话时最常用的循环：

from dapr.clients import DaprClient
from dapr.clients.grpc import conversation

@conversation.tool
def get_weather(location: str, unit: str = 'fahrenheit') -> str:
    return f"Weather in {location} (unit={unit})"

history: list[conversation.ConversationMessage] = [
    conversation.create_user_message("旧金山的天气怎么样？")]

with DaprClient() as client:
    # 第 1 轮
    resp1 = client.converse_alpha2(
        name="openai",
        inputs=[conversation.ConversationInputAlpha2(messages=history)],
        tools=conversation.get_registered_tools(),
        tool_choice='auto',
        temperature=1,
    )

    # 追加 assistant 消息；执行工具调用；再追加工具结果
    for msg in resp1.to_assistant_messages():
        history.append(msg)
        for tc in msg.of_assistant.tool_calls:
            # 重要：生产环境中必须校验输入并施加防护
            tool_output = conversation.execute_registered_tool(
                tc.function.name, tc.function.arguments
            )
            history.append(
                conversation.create_tool_message(
                    tool_id=tc.id, name=tc.function.name, content=str(tool_output)
                )
            )

    # 第 2 轮（LLM 会看到工具结果）
    history.append(conversation.create_user_message("我需要带伞吗？"))
    resp2 = client.converse_alpha2(
        name="openai",
        inputs=[conversation.ConversationInputAlpha2(messages=history)],
        tools=conversation.get_registered_tools(),
        temperature=1,
    )

    for msg in resp2.to_assistant_messages():
        history.append(msg)
        if not msg.of_assistant.tool_calls and msg.of_assistant.content:
            print(msg.of_assistant.content[0].text)

提示：

将函数用作工具：其他方式

当装饰器方式不方便时，还有两种选择。

A）从带类型的函数自动生成 schema：

from enum import Enum
from dapr.clients.grpc import conversation

class Units(Enum):
    CELSIUS = 'celsius'
    FAHRENHEIT = 'fahrenheit'

def get_weather(location: str, unit: Units = Units.FAHRENHEIT) -> str:
    return f"Weather in {location}"

fn = conversation.ConversationToolsFunction.from_function(get_weather)
weather_tool = conversation.ConversationTools(function=fn)

B）手写 JSON Schema（兜底方式）：

from dapr.clients.grpc import conversation

fn = conversation.ConversationToolsFunction(
    name='get_weather',
    description='Get current weather',
    parameters={
        'type': 'object',
        'properties': {
            'location': {'type': 'string'},
            'unit': {'type': 'string', 'enum': ['celsius', 'fahrenheit']},
        },
        'required': ['location'],
    },
)
weather_tool = conversation.ConversationTools(function=fn)

异步变体

按需使用异步客户端与异步工具执行辅助方法。

import asyncio
from dapr.aio.clients import DaprClient as AsyncDaprClient
from dapr.clients.grpc import conversation

@conversation.tool
def get_time() -> str:
    return '2025-01-01T12:00:00Z'

async def main():
    async with AsyncDaprClient() as client:
        msg = conversation.create_user_message('现在几点？')
        inp = conversation.ConversationInputAlpha2(messages=[msg])
        resp = await client.converse_alpha2(
            name='openai', inputs=[inp], tools=conversation.get_registered_tools()
        )
        for m in resp.to_assistant_messages():
            if m.of_assistant.content:
                print(m.of_assistant.content[0].text)

asyncio.run(main())

如果你需要异步执行工具（例如网络 I/O），请实现异步函数，并结合超时参数使用 conversation.execute_registered_tool_async。

安全与验证（必读）

LLM 可能会建议调用工具。所有由模型提供的参数都必须视为不可信输入。

建议：

另请参阅 dapr/clients/grpc/conversation.py 中的内联说明（如 tool()、ConversationTools、execute_registered_tool），了解参数绑定与错误处理细节。

关键辅助方法（速查）

本节汇总了示例中使用到的 dapr.clients.grpc.conversation 辅助工具。

提示：@conversation.tool 装饰器是创建工具最简单的方式。它会根据函数自动生成 schema，支持可选的命名空间或名称覆盖，并自动注册工具（若想延后注册，可设置 register=False）。