在 Python 中构建一个 Open Floor 鹦鹉代理
社区文章 发布于 2025 年 7 月 19 日
在本简短指南中，我们将使用 Python 共同构建一个简单的鹦鹉代理。鹦鹉代理将简单地重复您发送给它的所有内容，并在返回内容前面添加一个小的 🦜 表情符号。我们将借助 openfloor Python 包创建符合 Open Floor 协议的代理。
初始设置
首先，我们通过创建项目文件夹和安装所需包来设置项目
mkdir parrot-agent
cd parrot-agent
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
创建一个包含以下内容的 requirements.txt 文件
--extra-index-url https://test.pypi.org/simple/
events==0.5
jsonpath-ng>=1.5.0
openfloor
flask
然后安装依赖项
pip install -r requirements.txt
现在基本设置已完成，让我们开始一起编写代码！
第 1 步：构建鹦鹉代理类
让我们创建主代理文件。创建一个新文件 parrot_agent.py，它将包含我们代理的主要逻辑。
第 1.1 步：添加导入
让我们从 openfloor 包中导入所需的所有内容，将它们添加到 parrot_agent.py 文件的顶部
from openfloor import (
BotAgent,
Manifest,
Identification,
Capability,
SupportedLayers,
UtteranceEvent,
Envelope,
DialogEvent,
TextFeature,
To,
Sender,
PublishManifestsEvent,
Parameters
)
为什么要导入这些？
BotAgent- 我们将扩展的基类Manifest- 用于定义代理的能力和身份UtteranceEvent- 我们将处理文本消息的事件类型Envelope- Open Floor 消息的容器DialogEvent和TextFeature- 用于创建文本响应To和Sender- 用于消息寻址
第 1.2 步：启动 ParrotAgent 类
现在让我们通过扩展 BotAgent 来创建 ParrotAgent 类
class ParrotAgent(BotAgent):
"""
ParrotAgent - A simple agent that echoes back whatever it receives
Extends BotAgent to provide parrot functionality
"""
def __init__(self, manifest: Manifest):
super().__init__(manifest)
print(f"🦜 Parrot Agent initialized with speaker URI: {self.speakerUri}")
我们刚才做了什么
- 创建了一个扩展
BotAgent的类 - 添加了一个构造函数，它接受一个清单并将其传递给父类
- 清单将定义我们的代理可以做什么
第 1.3 步：重写语句处理程序
BotAgent 类提供了一个默认的 bot_on_utterance 方法，我们需要重写它。这就是奇迹发生的地方
def bot_on_utterance(self, event: UtteranceEvent, in_envelope: Envelope, out_envelope: Envelope) -> None:
"""
Override the utterance handler to provide parrot functionality
"""
print("🦜 Processing utterance event")
try:
# Extract the dialog event from the utterance parameters
dialog_event_data = event.parameters.get("dialogEvent")
if not dialog_event_data:
self._send_error_response("*chirp* I didn't receive a valid dialog event!", out_envelope)
return
# Convert to DialogEvent object if it's a dictionary
if isinstance(dialog_event_data, dict):
dialog_event = DialogEvent.from_dict(dialog_event_data)
elif isinstance(dialog_event_data, DialogEvent):
dialog_event = dialog_event_data
else:
self._send_error_response("*chirp* I didn't receive a valid dialog event!", out_envelope)
return
# Check if there's a text feature
text_feature = dialog_event.features.get("text")
if not text_feature or not hasattr(text_feature, 'tokens') or not text_feature.tokens:
self._send_error_response("*chirp* I can only repeat text messages!", out_envelope)
return
# Extract the original text from tokens
original_text = "".join(token.value for token in text_feature.tokens if token.value)
# Create parrot response with emoji prefix
parrot_text = f"🦜 {original_text}"
# Create the response dialog event
response_dialog = DialogEvent(
speakerUri=self.speakerUri,
features={"text": TextFeature(values=[parrot_text])}
)
# Create and add the utterance event to the response
response_utterance = UtteranceEvent(
dialogEvent=response_dialog,
to=To(speakerUri=in_envelope.sender.speakerUri)
)
out_envelope.events.append(response_utterance)
print(f"🦜 Echoing back: {parrot_text}")
except Exception as error:
print(f"🦜 Error in parrot utterance handling: {error}")
self._send_error_response("*confused chirp* Something went wrong while trying to repeat that!", out_envelope)
鹦鹉逻辑
- 从语句参数中提取对话事件
- 获取文本特征并从标记中提取文本
- 添加 🦜 表情符号前缀
- 创建对话事件响应
- 将其发送回原始发件人
第 1.4 步：添加辅助方法
让我们添加在 bot_on_utterance 中调用的辅助方法
def _send_error_response(self, message: str, out_envelope: Envelope) -> None:
"""Helper method to send error responses"""
error_dialog = DialogEvent(
speakerUri=self.speakerUri,
features={"text": TextFeature(values=[message])}
)
error_utterance = UtteranceEvent(
dialogEvent=error_dialog
)
out_envelope.events.append(error_utterance)
第 1.5 步：重写清单处理程序
我们还需要正确处理清单请求
def bot_on_get_manifests(self, event, in_envelope: Envelope, out_envelope: Envelope) -> None:
"""
Handle manifest requests by sending our capabilities
"""
print("🦜 Sending manifest information")
# Create the publish manifests response
publish_event = PublishManifestsEvent(
parameters=Parameters({
"servicingManifests": [self._manifest],
"discoveryManifests": []
}),
to=To(speakerUri=in_envelope.sender.speakerUri)
)
out_envelope.events.append(publish_event)
第 1.6 步：添加工厂函数
在类之后，添加此带有默认配置的工厂函数
def create_parrot_agent(
speaker_uri: str,
service_url: str,
name: str = "Parrot Agent",
organization: str = "OpenFloor Demo",
description: str = "A simple parrot agent that echoes back messages with a 🦜 emoji"
) -> ParrotAgent:
"""
Factory function to create a ParrotAgent with default configuration
"""
# Create the identification
identification = Identification(
speakerUri=speaker_uri,
serviceUrl=service_url,
organization=organization,
conversationalName=name,
synopsis=description
)
# Create the capabilities
capability = Capability(
keyphrases=['echo', 'repeat', 'parrot', 'say'],
descriptions=[
'Echoes back any text message with a 🦜 emoji',
'Repeats user input verbatim',
'Simple text mirroring functionality'
],
supportedLayers=SupportedLayers(
input=["text"],
output=["text"]
)
)
# Create the manifest
manifest = Manifest(
identification=identification,
capabilities=[capability]
)
return ParrotAgent(manifest)
此工厂的作用
- 接受一些带有默认值的配置选项
- 创建描述我们代理的标识对象
- 定义告诉其他人我们的代理可以做什么的功能
- 创建一个结合了标识和功能的清单
- 返回一个新的 ParrotAgent 实例
第 2 步：构建 Flask 服务器
代理本身已经完成，但如何与它对话呢？我们需要为此构建 Flask 服务器，所以从创建 server.py 文件开始。
第 2.1 步：添加导入
在顶部添加这些导入
from flask import Flask, request, jsonify
from parrot_agent import create_parrot_agent
from openfloor import Envelope, Payload
import json
import os
第 2.2 步：创建 Flask 应用程序
app = Flask(__name__)
# Configure CORS for specific origin
@app.after_request
def after_request(response):
allowed_origin = 'http://127.0.0.1:4000'
origin = request.headers.get('Origin')
if origin == allowed_origin:
response.headers.add('Access-Control-Allow-Origin', allowed_origin)
response.headers.add('Access-Control-Allow-Methods', 'POST, OPTIONS')
response.headers.add('Access-Control-Allow-Headers', 'Content-Type')
return response
@app.route('/', methods=['OPTIONS'])
def handle_options():
"""Handle preflight OPTIONS requests"""
return '', 200
为什么需要 CORS 设置？
- 只允许来自特定域的请求
- 处理预检
OPTIONS请求 - 限制为
POST方法和Content-Type头部
第 2.3 步：创建代理实例
现在我们需要使用工厂函数 create_parrot_agent 来创建我们的鹦鹉。
# Create the parrot agent instance
parrot_agent = create_parrot_agent(
speaker_uri='tag:openfloor-demo.com,2025:parrot-agent',
service_url=os.environ.get('SERVICE_URL', 'https://:8080/'),
name='Polly the Parrot',
organization='OpenFloor Demo Corp',
description='A friendly parrot that repeats everything you say!'
)
print(f"🦜 Parrot agent created: {parrot_agent.speakerUri}")
第 2.4 步：逐步构建主端点
现在我们有了代理和 Flask 应用程序，但最重要的部分仍然缺失，那就是我们的端点
@app.route('/', methods=['POST'])
def handle_openfloor_message():
"""Main Open Floor Protocol endpoint"""
try:
print(f"🦜 Received request: {json.dumps(request.json, indent=2)}")
# Validate the incoming payload
if not request.json:
return jsonify({
'error': 'Invalid JSON payload'
}), 400
# Parse the payload
try:
if 'openFloor' in request.json:
# Direct payload format
payload = Payload.from_dict(request.json)
incoming_envelope = payload.openFloor
else:
# Direct envelope format
incoming_envelope = Envelope.from_dict(request.json)
except Exception as parse_error:
print(f"🦜 Parsing error: {parse_error}")
return jsonify({
'error': 'Invalid OpenFloor payload format',
'details': str(parse_error)
}), 400
print(f"🦜 Processing envelope from: {incoming_envelope.sender.speakerUri}")
# Process the envelope through the parrot agent
outgoing_envelope = parrot_agent.process_envelope(incoming_envelope)
# Create response payload
response_payload = Payload(openFloor=outgoing_envelope)
response = dict(response_payload)
print(f"🦜 Sending response: {json.dumps(response, indent=2, default=str)}")
return jsonify(response)
except Exception as error:
print(f"🦜 Error processing request: {error}")
return jsonify({
'error': 'Internal server error',
'message': str(error)
}), 500
这里发生了什么
- 验证我们收到的是有效的 JSON
- 将有效载荷解析为 Open Floor 信封
- 通过我们的鹦鹉代理处理它
- 创建并发送格式正确的响应有效载荷
第 3 步：创建入口点
我们最后创建了一个简单的 main.py 作为我们的入口点
from server import app
import os
if __name__ == '__main__':
port = int(os.environ.get('PORT', 8080))
print(f"🦜 Parrot Agent server starting on port {port}")
app.run(host='0.0.0.0', port=port, debug=True)
第 4 步：最终设置
您的项目结构现在应该看起来像这样
parrot-agent/
├── requirements.txt
├── parrot_agent.py
├── server.py
└── main.py
测试您的实现
运行此命令进行测试
python main.py
将您的清单或语句请求发送到 https://:8080/，看看它是否正常工作！您还可以下载简单的单 HTML 文件清单和语句聊天 azettl/openfloor-js-chat 来在本地测试您的代理。
如果您觉得本指南有用，请关注我以获取更多信息，并在评论中告诉我您用它构建了什么！