Universal Conversation Orchestration Middleware

通用对话编排中间件，提供多Agent协作框架，支持状态机驱动的Agent编排、工具调用、记忆管理、流程控制、流式响应、会话管理等功能，可嵌入任意Java/Spring Boot或Python应用。

功能特性

核心功能

• 状态机驱动的Agent编排：支持JSON/YAML配置、条件表达式、状态嵌套
• 工具调用系统：支持工具描述、工具链、异步工具、工具权限
• 智能记忆管理：支持长期记忆、记忆检索、记忆摘要、多维度记忆
• 增强AI能力：支持多模型集成、提示工程、上下文压缩、情感分析
• 性能优化：支持多级缓存、异步处理、资源管理
• 插件系统：支持插件加载、钩子机制、扩展点
• 错误处理和监控：支持统一错误处理、详细日志记录、指标监控
• 安全性增强：支持输入验证、输入净化、权限控制
• 容器化和云服务集成：支持Docker容器化、Redis集成

快速开始

安装依赖

pip install -r requirements.txt

配置环境变量

创建 .env 文件并配置以下环境变量：

# OpenAI API Key
OPENAI_API_KEY=your_openai_api_key_here

# Redis Configuration
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_PASSWORD=

# Application Configuration
SESSION_TIMEOUT=3600
MAX_HISTORY_LENGTH=100

基本使用

from core.api.api import ConversationOrchestrationAPI

# 初始化API
api = ConversationOrchestrationAPI()

# 注册工具
def get_weather(city):
    """获取城市天气"""
    return f"{city}的天气晴朗，温度25°C"

api.register_tool(
    tool_name="get_weather",
    tool_function=get_weather,
    description="获取指定城市的天气信息",
    params={"city": "城市名称"},
    permissions=["weather:read"]
)

# 处理消息
user_id = "user123"
session_id = "session456"
message = "北京的天气怎么样？"
response = api.process_message(user_id, message, session_id)
print(response["response"])

高级使用

异步处理

import asyncio

async def main():
    api = ConversationOrchestrationAPI()
    response = await api.process_message_async(user_id, message, session_id)
    print(response["response"])

asyncio.run(main())

流式响应

def callback(chunk):
    print(chunk, end="", flush=True)

api.stream_message(user_id, message, session_id, callback)

插件使用

# 加载插件
api.load_plugin("path/to/plugin.py")

# 列出所有插件
plugins = api.list_plugins()
print(plugins)

容器化部署

使用Docker

# 构建镜像
docker build -t conversation-orchestration-middleware .

# 运行容器
docker run -p 8000:8000 --env-file .env conversation-orchestration-middleware

使用Docker Compose

docker-compose up -d

目录结构

agent/
├── core/
│   ├── agents/            # 智能体模块
│   ├── config/            # 配置模块
│   ├── engine/            # 核心引擎
│   ├── api/               # API接口
│   ├── plugins/           # 插件系统
│   └── utils/             # 工具模块
├── examples/              # 示例项目
├── Dockerfile             # Docker构建文件
├── docker-compose.yml     # Docker Compose配置
├── requirements.txt       # 依赖文件
├── .env                   # 环境变量
└── README.md              # 项目文档

API 文档

核心API

• process_message(user_id, message, session_id, user_permissions=None, topics=None, model_type="default", template="default") - 处理消息
• process_message_async(user_id, message, session_id, user_permissions=None, topics=None, model_type="default", template="default") - 异步处理消息
• stream_message(user_id, message, session_id, callback, model_type="default", template="default") - 流式处理消息
• register_tool(tool_name, tool_function, description="", params=None, permissions=None) - 注册工具
• register_prompt_template(name, template) - 注册提示模板
• register_model(model_type, model_name) - 注册模型
• analyze_sentiment(text) - 分析文本情感
• summarize(text, max_length=100) - 文本摘要
• load_plugin(plugin_path) - 加载插件
• list_plugins() - 列出所有插件
• clear_memory(user_id, session_id) - 清除记忆

插件开发

创建插件

from core.plugins.base_plugin import BasePlugin

class MyPlugin(BasePlugin):
    def __init__(self):
        super().__init__()
        self.description = "我的插件"
    
    def initialize(self, config):
        print("插件初始化")
        return True
    
    def shutdown(self):
        print("插件关闭")
        return True
    
    def on_message_received(self, user_id, message, session_id):
        print(f"收到消息: {message}")
        return None

加载插件

api.load_plugin("path/to/my_plugin.py")

贡献

欢迎贡献代码、报告问题或提出建议！

许可证

MIT License