Skip to content

API.md

Latest commit

 

History

History
742 lines (557 loc) · 13.3 KB

API.md

File metadata and controls

742 lines (557 loc) · 13.3 KB

API接口文档

目录

核心类API

Orchestrator - 回复编排器

初始化

from app.orchestrator import Orchestrator
from agent.llm_client import LLMClient
from memory.context_manager import ChatContextManager

# 创建LLM客户端
llm_client = LLMClient(
    client=openai_client,
    model="qwen-max",
    temperature=0.3,
    max_tokens=192
)

# 创建上下文管理器
context_manager = ChatContextManager(
    recent_window=6,
    summary_max_chars=140,
    facts_max_chars=180
)

# 创建编排器
orchestrator = Orchestrator(
    llm_client=llm_client,
    context_manager=context_manager,
    use_summary=True  # 启用摘要优化
)

reply() - 生成回复

功能：生成回复消息

参数：

  • chat_id (str): 会话ID
  • user_text (str): 用户消息
  • item_info (Dict[str, Any]): 商品信息

返回：

{
    "reply": str,      # 回复内容
    "intent": str      # 意图类型
}

示例：

result = orchestrator.reply(
    chat_id="58980844020",
    user_text="这个多少钱？",
    item_info={
        "title": "iPhone 15 Pro Max",
        "price": "8999",
        "desc": "全新原装..."
    }
)

print(result["reply"])   # "您好，价格是8999元"
print(result["intent"])  # "price"

XianyuLive - WebSocket主循环

初始化

from app.xianyu_live import XianyuLive

live = XianyuLive(
    cookies_str="your_cookies_here",
    orchestrator=orchestrator
)

run() - 启动主循环

功能：启动WebSocket主循环

参数：无

返回：无（异步方法）

示例：

import asyncio

async def main():
    await live.run()

asyncio.run(main())

toggle_manual_mode() - 切换人工模式

功能：切换人工模式

参数：

  • chat_id (str): 会话ID

返回：

  • str: "auto" 或 "manual"

示例：

mode = live.toggle_manual_mode("58980844020")
print(mode)  # "manual" 或 "auto"

is_manual_mode() - 检查人工模式

功能：检查是否处于人工模式

参数：

  • chat_id (str): 会话ID

返回：

  • bool: 是否处于人工模式

示例：

is_manual = live.is_manual_mode("58980844020")
print(is_manual)  # True 或 False

RuleEngine - 规则引擎

初始化

from app.rule_engine import RuleEngine

rules = RuleEngine("config/rules.json")

decide() - 规则决策

功能：根据规则做出决策

参数：

  • user_text (str): 用户消息
  • item_compact_json (str): 商品摘要JSON

返回：

RuleDecision(
    action=str,      # 动作类型
    rule_id=str,     # 规则ID
    intent=str,      # 意图（route动作）
    reply=str        # 回复（reply动作）
)

示例：

decision = rules.decide("多少钱？", "{}")

if decision.action == "reply":
    print(decision.reply)  # 直接回复
elif decision.action == "route":
    print(decision.intent)  # 路由到指定意图

reload() - 热加载规则

功能：重新加载规则文件

参数：无

返回：无

示例：

rules.reload()  # 重新加载config/rules.json

add_rule() - 添加规则

功能：添加新规则

参数：

  • rule (Dict[str, Any]): 规则配置

返回：无

示例：

rules.add_rule({
    "id": "custom_rule",
    "name": "自定义规则",
    "match": {
        "type": "contains",
        "values": ["关键词"]
    },
    "action": "reply",
    "reply": "自定义回复"
})

remove_rule() - 移除规则

功能：移除规则

参数：

  • rule_name (str): 规则名称

返回：

  • bool: 是否成功移除

示例：

success = rules.remove_rule("custom_rule")
print(success)  # True 或 False

LLMClient - LLM客户端

初始化

from agent.llm_client import LLMClient
from openai import OpenAI

client = OpenAI(api_key="your_api_key")
llm_client = LLMClient(
    client=client,
    model="qwen-max",
    temperature=0.3,
    max_tokens=192
)

chat() - 发送聊天请求

功能：发送聊天请求到LLM

参数：

  • messages (List[Dict[str, str]]): 消息列表
  • max_tokens (Optional[int]): 最大token数
  • meta (Optional[dict]): 元数据

返回：

  • str: 回复内容

示例：

messages = [
    {"role": "system", "content": "你是客服"},
    {"role": "user", "content": "你好"}
]

meta = {"intent": "default"}
reply = llm_client.chat(messages, meta=meta)

print(reply)  # "你好，有什么可以帮你的？"
print(meta)   # 包含token使用统计

Router - 意图路由

初始化

from app.orchestrator import Router

router = Router(
    llm_client=llm_client,
    router_prompt="把用户消息分类为..."
)

classify() - 分类意图

功能：分类用户意图

参数：

  • user_text (str): 用户消息
  • item_compact_json (str): 商品摘要JSON

返回：

  • str: 意图类型（price/tech/default/no_reply）

示例：

intent = router.classify("这个多少钱？", "{}")
print(intent)  # "price"

Summarizer - 摘要生成

ConversationSummarizer - 对话摘要

初始化：

from agent.summarizer import ConversationSummarizer

summarizer = ConversationSummarizer(max_summary_chars=150)

summarize_conversation() - 摘要对话：

messages = [
    {"role": "user", "content": "你好"},
    {"role": "assistant", "content": "你好"}
]

summary = summarizer.summarize_conversation(messages)
print(summary)
# {
#     "turns": 1,
#     "user_topics": [],
#     "price_mentions": 0,
#     "tech_mentions": 0,
#     "summary": "对话1轮"
# }

format_structured_summary() - 格式化摘要：

formatted = summarizer.format_structured_summary(summary)
print(formatted)  # "对话1轮"

ItemSummarizer - 商品摘要

初始化：

from agent.summarizer import ItemSummarizer

summarizer = ItemSummarizer(max_chars=120)

summarize_item() - 摘要商品：

item_info = {
    "type": "physical",
    "title": "iPhone 15",
    "price": "8999"
}

summary = summarizer.summarize_item(item_info)
print(summary)
# {
#     "type": "physical",
#     "title": "iPhone 15",
#     "price": "8999",
#     "key_features": [],
#     "summary": "physical ¥8999 iPhone 15"
# }

ContextManager - 上下文管理

初始化

from memory.context_manager import ChatContextManager

context_manager = ChatContextManager(
    recent_window=6,
    summary_max_chars=140,
    facts_max_chars=180
)

add_message_by_chat() - 添加消息

功能：添加消息到上下文

参数：

  • chat_id (str): 会话ID
  • user_id (str): 用户ID
  • item_id (str): 商品ID
  • role (str): 角色（user/assistant）
  • content (str): 消息内容

返回：无

示例：

context_manager.add_message_by_chat(
    chat_id="58980844020",
    user_id="2239385583",
    item_id="1024961377690",
    role="user",
    content="你好"
)

get_context_by_chat() - 获取上下文

功能：获取会话上下文

参数：

  • chat_id (str): 会话ID
  • compact (bool): 是否压缩格式

返回：

  • List[Dict[str, str]]: 消息列表

示例：

messages = context_manager.get_context_by_chat(
    chat_id="58980844020",
    compact=True
)

for msg in messages:
    print(f"{msg['role']}: {msg['content']}")

配置API

环境变量配置

import os
from dotenv import load_dotenv

load_dotenv()

# API配置
api_key = os.getenv("API_KEY")
base_url = os.getenv("BASE_URL")
model_name = os.getenv("MODEL_NAME")

# 闲鱼配置
cookies_str = os.getenv("COOKIES_STR")

# 可选配置
log_level = os.getenv("LOG_LEVEL", "INFO")
heartbeat_interval = int(os.getenv("HEARTBEAT_INTERVAL", "15"))
token_refresh_interval = int(os.getenv("TOKEN_REFRESH_INTERVAL", "3600"))
use_summary = os.getenv("USE_SUMMARY", "True").lower() == "true"

规则配置

import json

# 加载规则
with open("config/rules.json", "r", encoding="utf-8") as f:
    rules_config = json.load(f)

# 规则结构
rule = {
    "id": "rule_id",
    "name": "规则名称",
    "item_type": "any",  # any/physical/virtual
    "match": {
        "type": "contains",  # contains/equals/regex/length
        "values": ["关键词"]
    },
    "action": "reply",  # block/no_reply/reply/route
    "reply": "回复内容",  # reply动作时使用
    "intent": "price"   # route动作时使用
}

Prompt配置

import json

# 加载Prompt
with open("config/prompts.json", "r", encoding="utf-8") as f:
    prompts = json.load(f)

# Prompt结构
prompts = {
    "system_global": "全局系统提示",
    "router": "路由提示",
    "responder": {
        "price": "价格Agent提示",
        "tech": "技术Agent提示",
        "default": "默认Agent提示"
    }
}

工具API

日志工具

from loguru import logger

# 配置日志
import sys
logger.remove()
logger.add(
    sys.stderr,
    level="INFO",
    format="<green>{time}</green> | <level>{level}</level> | <cyan>{message}</cyan>"
)

# 使用日志
logger.info("信息日志")
logger.warning("警告日志")
logger.error("错误日志")
logger.debug("调试日志")

闲鱼工具

from utils.xianyu_utils import (
    generate_mid,
    generate_uuid,
    trans_cookies,
    generate_device_id,
    decrypt
)

# 生成消息ID
mid = generate_mid()

# 生成UUID
uuid = generate_uuid()

# 转换Cookie
cookies = trans_cookies("cookie_string")

# 生成设备ID
device_id = generate_device_id("user_id")

# 解密消息
decrypted = decrypt("encrypted_data")

数据结构

RuleDecision - 规则决策

from dataclasses import dataclass
from typing import Optional

@dataclass
class RuleDecision:
    action: str                    # block/no_reply/reply/route/none
    rule_id: Optional[str] = None  # 规则ID
    intent: Optional[str] = None   # 意图（route动作）
    reply: Optional[str] = None    # 回复（reply动作）

消息格式

message = {
    "role": "user",        # user/assistant/system
    "content": "消息内容"
}

商品信息

item_info = {
    "type": "physical",    # physical/virtual
    "title": "商品标题",
    "price": "价格",
    "desc": "商品描述",
    "props": {             # 商品属性
        "品牌": "品牌名",
        "型号": "型号名"
    }
}

统计数据

stats = {
    "llm_calls": 0,        # LLM调用次数
    "rule_hits": 0,        # 规则命中次数
    "by_intent": {         # 按意图统计
        "price": {
            "calls": 0,
            "prompt_chars": 0,
            "completion_chars": 0,
            "tokens": 0
        }
    }
}

错误处理

异常类型

# API错误
class APIError(Exception):
    pass

# 配置错误
class ConfigError(Exception):
    pass

# 网络错误
class NetworkError(Exception):
    pass

错误处理示例

try:
    result = orchestrator.reply(chat_id, user_text, item_info)
except APIError as e:
    logger.error(f"API错误: {e}")
    result = {"reply": "系统繁忙，请稍后再试", "intent": "error"}
except Exception as e:
    logger.error(f"未知错误: {e}")
    result = {"reply": "系统错误", "intent": "error"}

使用示例

完整示例

import asyncio
from openai import OpenAI
from app.xianyu_live import XianyuLive
from app.orchestrator import Orchestrator
from agent.llm_client import LLMClient
from memory.context_manager import ChatContextManager

async def main():
    # 创建LLM客户端
    openai_client = OpenAI(
        api_key="your_api_key",
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
    )
    
    llm_client = LLMClient(
        client=openai_client,
        model="qwen-max"
    )
    
    # 创建上下文管理器
    context_manager = ChatContextManager()
    
    # 创建编排器
    orchestrator = Orchestrator(
        llm_client=llm_client,
        context_manager=context_manager
    )
    
    # 创建并启动
    live = XianyuLive(
        cookies_str="your_cookies",
        orchestrator=orchestrator
    )
    
    await live.run()

if __name__ == "__main__":
    asyncio.run(main())

性能优化建议

1. 启用摘要优化

orchestrator = Orchestrator(
    llm_client=llm_client,
    context_manager=context_manager,
    use_summary=True  # 节省50-70% token
)

2. 调整上下文窗口

context_manager = ChatContextManager(
    recent_window=6,        # 保留最近6轮对话
    summary_max_chars=140,  # 摘要最大字符数
    facts_max_chars=180     # 事实最大字符数
)

3. 优化规则配置

# 高频规则放在前面
rules = [
    {"id": "high_freq_rule", ...},  # 高频规则
    {"id": "medium_freq_rule", ...}, # 中频规则
    {"id": "low_freq_rule", ...}    # 低频规则
]

总结

本文档提供了XianyuAutoAgent的完整API接口说明，包括核心类API、配置API、工具API和数据结构。通过这些API，您可以灵活地使用和扩展系统功能。