mirror of
https://github.com/zhayujie/chatgpt-on-wechat.git
synced 2026-06-02 09:48:22 +08:00
Compare commits
7 Commits
feat-web-c
...
feat-multi
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7cce224499 | ||
|
|
925d728a86 | ||
|
|
9917552b4b | ||
|
|
29bfbecdc9 | ||
|
|
1a7a8c98d9 | ||
|
|
cddb38ac3d | ||
|
|
d610608391 |
@@ -24,8 +24,9 @@
|
||||
|
||||
## 声明
|
||||
|
||||
1. 本项目遵循 [MIT开源协议](/LICENSE),主要用于技术研究和学习,使用本项目时需遵守所在地法律法规、相关政策以及企业章程,禁止用于任何违法或侵犯他人权益的行为。任何个人、团队和企业,无论以何种方式使用该项目、对何对象提供服务,所产生的一切后果,本项目均不承担任何责任
|
||||
2. 成本与安全:Agent模式下Token使用量高于普通对话模式,请根据效果及成本综合选择模型。Agent具有访问所在操作系统的能力,请谨慎选择项目部署环境。同时项目也会持续升级安全机制、并降低模型消耗成本
|
||||
1. 本项目遵循 [MIT开源协议](/LICENSE),主要用于技术研究和学习,使用本项目时需遵守所在地法律法规、相关政策以及企业章程,禁止用于任何违法或侵犯他人权益的行为。任何个人、团队和企业,无论以何种方式使用该项目、对何对象提供服务,所产生的一切后果,本项目均不承担任何责任。
|
||||
2. 成本与安全:Agent模式下Token使用量高于普通对话模式,请根据效果及成本综合选择模型。Agent具有访问所在操作系统的能力,请谨慎选择项目部署环境。同时项目也会持续升级安全机制、并降低模型消耗成本。
|
||||
3. CowAgent项目专注于开源技术开发,不会参与、授权或发行任何加密货币。
|
||||
|
||||
## 演示
|
||||
|
||||
@@ -607,10 +608,12 @@ API Key创建:在 [控制台](https://aistudio.google.com/app/apikey?hl=zh-cn)
|
||||
|
||||
以下对可接入通道的配置方式进行说明,应用通道代码在项目的 `channel/` 目录下。
|
||||
|
||||
支持同时可接入多个通道,配置时可通过逗号进行分割,例如 `"channel_type": "feishu,dingtalk"`。
|
||||
|
||||
<details>
|
||||
<summary>1. Web</summary>
|
||||
|
||||
项目启动后默认运行Web通道,配置如下:
|
||||
项目启动后会默认运行Web控制台,配置如下:
|
||||
|
||||
```json
|
||||
{
|
||||
|
||||
@@ -158,6 +158,7 @@ class ChatService:
|
||||
logger.info(f"[ChatService] Agent run completed: session={session_id}")
|
||||
|
||||
|
||||
|
||||
class _StreamState:
|
||||
"""Mutable state shared between the event callback and the run method."""
|
||||
|
||||
|
||||
@@ -1,11 +1,21 @@
|
||||
"""
|
||||
Memory module for AgentMesh
|
||||
|
||||
Provides long-term memory capabilities with hybrid search (vector + keyword)
|
||||
Provides both long-term memory (vector/keyword search) and short-term
|
||||
conversation history persistence (SQLite).
|
||||
"""
|
||||
|
||||
from agent.memory.manager import MemoryManager
|
||||
from agent.memory.config import MemoryConfig, get_default_memory_config, set_global_memory_config
|
||||
from agent.memory.embedding import create_embedding_provider
|
||||
from agent.memory.conversation_store import ConversationStore, get_conversation_store
|
||||
|
||||
__all__ = ['MemoryManager', 'MemoryConfig', 'get_default_memory_config', 'set_global_memory_config', 'create_embedding_provider']
|
||||
__all__ = [
|
||||
'MemoryManager',
|
||||
'MemoryConfig',
|
||||
'get_default_memory_config',
|
||||
'set_global_memory_config',
|
||||
'create_embedding_provider',
|
||||
'ConversationStore',
|
||||
'get_conversation_store',
|
||||
]
|
||||
|
||||
618
agent/memory/conversation_store.py
Normal file
618
agent/memory/conversation_store.py
Normal file
@@ -0,0 +1,618 @@
|
||||
"""
|
||||
Conversation history persistence using SQLite.
|
||||
|
||||
Design:
|
||||
- sessions table: per-session metadata (channel_type, last_active, msg_count)
|
||||
- messages table: individual messages stored as JSON, append-only
|
||||
- Pruning: age-based only (sessions not updated within N days are deleted)
|
||||
- Thread-safe via a single in-process lock
|
||||
|
||||
Storage path: ~/cow/sessions/conversations.db
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sqlite3
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from common.log import logger
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Schema
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_DDL = """
|
||||
CREATE TABLE IF NOT EXISTS sessions (
|
||||
session_id TEXT PRIMARY KEY,
|
||||
channel_type TEXT NOT NULL DEFAULT '',
|
||||
created_at INTEGER NOT NULL,
|
||||
last_active INTEGER NOT NULL,
|
||||
msg_count INTEGER NOT NULL DEFAULT 0
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS messages (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
session_id TEXT NOT NULL,
|
||||
seq INTEGER NOT NULL,
|
||||
role TEXT NOT NULL,
|
||||
content TEXT NOT NULL,
|
||||
created_at INTEGER NOT NULL,
|
||||
UNIQUE (session_id, seq)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_messages_session
|
||||
ON messages (session_id, seq);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_sessions_last_active
|
||||
ON sessions (last_active);
|
||||
"""
|
||||
|
||||
# Migration: add channel_type column to existing databases that predate it.
|
||||
_MIGRATION_ADD_CHANNEL_TYPE = """
|
||||
ALTER TABLE sessions ADD COLUMN channel_type TEXT NOT NULL DEFAULT '';
|
||||
"""
|
||||
|
||||
DEFAULT_MAX_AGE_DAYS: int = 30
|
||||
|
||||
|
||||
def _is_visible_user_message(content: Any) -> bool:
|
||||
"""
|
||||
Return True when a user-role message represents actual user input
|
||||
(not an internal tool_result injected by the agent loop).
|
||||
"""
|
||||
if isinstance(content, str):
|
||||
return bool(content.strip())
|
||||
if isinstance(content, list):
|
||||
return any(
|
||||
isinstance(b, dict) and b.get("type") == "text"
|
||||
for b in content
|
||||
)
|
||||
return False
|
||||
|
||||
|
||||
def _extract_display_text(content: Any) -> str:
|
||||
"""
|
||||
Extract the human-readable text portion from a message content value.
|
||||
Returns an empty string for tool_use / tool_result blocks.
|
||||
"""
|
||||
if isinstance(content, str):
|
||||
return content.strip()
|
||||
if isinstance(content, list):
|
||||
parts = [
|
||||
b.get("text", "")
|
||||
for b in content
|
||||
if isinstance(b, dict) and b.get("type") == "text"
|
||||
]
|
||||
return "\n".join(p for p in parts if p).strip()
|
||||
return ""
|
||||
|
||||
|
||||
def _extract_tool_calls(content: Any) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Extract tool_use blocks from an assistant message content.
|
||||
Returns a list of {name, arguments} dicts (result filled in later).
|
||||
"""
|
||||
if not isinstance(content, list):
|
||||
return []
|
||||
return [
|
||||
{"id": b.get("id", ""), "name": b.get("name", ""), "arguments": b.get("input", {})}
|
||||
for b in content
|
||||
if isinstance(b, dict) and b.get("type") == "tool_use"
|
||||
]
|
||||
|
||||
|
||||
def _extract_tool_results(content: Any) -> Dict[str, str]:
|
||||
"""
|
||||
Extract tool_result blocks from a user message, keyed by tool_use_id.
|
||||
"""
|
||||
if not isinstance(content, list):
|
||||
return {}
|
||||
results = {}
|
||||
for b in content:
|
||||
if not isinstance(b, dict) or b.get("type") != "tool_result":
|
||||
continue
|
||||
tool_id = b.get("tool_use_id", "")
|
||||
result_content = b.get("content", "")
|
||||
if isinstance(result_content, list):
|
||||
result_content = "\n".join(
|
||||
rb.get("text", "") for rb in result_content
|
||||
if isinstance(rb, dict) and rb.get("type") == "text"
|
||||
)
|
||||
results[tool_id] = str(result_content)
|
||||
return results
|
||||
|
||||
|
||||
def _group_into_display_turns(
|
||||
rows: List[tuple],
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Convert raw (role, content_json, created_at) DB rows into display turns.
|
||||
|
||||
One display turn = one visible user message + one merged assistant reply.
|
||||
All intermediate assistant messages (those carrying tool_use) and the final
|
||||
assistant text reply produced for the same user query are collapsed into a
|
||||
single assistant turn, exactly matching the live SSE rendering where tools
|
||||
and the final answer appear inside the same bubble.
|
||||
|
||||
Grouping rules:
|
||||
- A visible user message starts a new group.
|
||||
- tool_result user messages are internal; their content is attached to the
|
||||
matching tool_use entry via tool_use_id and they never become own turns.
|
||||
- All assistant messages within a group are merged:
|
||||
* tool_use blocks → tool_calls list (result filled from tool_results)
|
||||
* text blocks → last non-empty text becomes the display content
|
||||
"""
|
||||
# ------------------------------------------------------------------ #
|
||||
# Pass 1: split rows into groups, each starting with a visible user msg
|
||||
# ------------------------------------------------------------------ #
|
||||
# group = (user_row | None, [subsequent_rows])
|
||||
# user_row: (content, created_at)
|
||||
groups: List[tuple] = []
|
||||
cur_user: Optional[tuple] = None
|
||||
cur_rest: List[tuple] = []
|
||||
started = False
|
||||
|
||||
for role, raw_content, created_at in rows:
|
||||
try:
|
||||
content = json.loads(raw_content)
|
||||
except Exception:
|
||||
content = raw_content
|
||||
|
||||
if role == "user" and _is_visible_user_message(content):
|
||||
if started:
|
||||
groups.append((cur_user, cur_rest))
|
||||
cur_user = (content, created_at)
|
||||
cur_rest = []
|
||||
started = True
|
||||
else:
|
||||
cur_rest.append((role, content, created_at))
|
||||
|
||||
if started:
|
||||
groups.append((cur_user, cur_rest))
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# Pass 2: build display turns from each group
|
||||
# ------------------------------------------------------------------ #
|
||||
turns: List[Dict[str, Any]] = []
|
||||
|
||||
for user_row, rest in groups:
|
||||
# User turn
|
||||
if user_row:
|
||||
content, created_at = user_row
|
||||
text = _extract_display_text(content)
|
||||
if text:
|
||||
turns.append({"role": "user", "content": text, "created_at": created_at})
|
||||
|
||||
# Collect all tool_calls and tool_results from the rest of the group
|
||||
all_tool_calls: List[Dict[str, Any]] = []
|
||||
tool_results: Dict[str, str] = {}
|
||||
final_text = ""
|
||||
final_ts: Optional[int] = None
|
||||
|
||||
for role, content, created_at in rest:
|
||||
if role == "user":
|
||||
tool_results.update(_extract_tool_results(content))
|
||||
elif role == "assistant":
|
||||
tcs = _extract_tool_calls(content)
|
||||
all_tool_calls.extend(tcs)
|
||||
t = _extract_display_text(content)
|
||||
if t:
|
||||
final_text = t
|
||||
final_ts = created_at
|
||||
|
||||
# Attach tool results to their matching tool_call entries
|
||||
for tc in all_tool_calls:
|
||||
tc["result"] = tool_results.get(tc.get("id", ""), "")
|
||||
|
||||
if final_text or all_tool_calls:
|
||||
turns.append({
|
||||
"role": "assistant",
|
||||
"content": final_text,
|
||||
"tool_calls": all_tool_calls,
|
||||
"created_at": final_ts or (user_row[1] if user_row else 0),
|
||||
})
|
||||
|
||||
return turns
|
||||
|
||||
|
||||
class ConversationStore:
|
||||
"""
|
||||
SQLite-backed store for per-session conversation history.
|
||||
|
||||
Usage:
|
||||
store = ConversationStore(db_path)
|
||||
store.append_messages("user_123", new_messages, channel_type="feishu")
|
||||
msgs = store.load_messages("user_123", max_turns=30)
|
||||
"""
|
||||
|
||||
def __init__(self, db_path: Path):
|
||||
self._db_path = db_path
|
||||
self._lock = threading.Lock()
|
||||
self._init_db()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Public API
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def load_messages(
|
||||
self,
|
||||
session_id: str,
|
||||
max_turns: int = 30,
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Load the most recent messages for a session, for injection into the LLM.
|
||||
|
||||
ALL message types (user text, assistant tool_use, tool_result) are returned
|
||||
in their original JSON form so the LLM can reconstruct the full context.
|
||||
|
||||
max_turns is a *visible-turn* count: we count only user messages whose
|
||||
content is actual user text (not tool_result blocks). This prevents
|
||||
tool-heavy sessions from exhausting the turn budget prematurely.
|
||||
|
||||
Args:
|
||||
session_id: Unique session identifier.
|
||||
max_turns: Maximum number of visible user-assistant turns to keep.
|
||||
|
||||
Returns:
|
||||
Chronologically ordered list of message dicts (role, content).
|
||||
"""
|
||||
with self._lock:
|
||||
conn = self._connect()
|
||||
try:
|
||||
rows = conn.execute(
|
||||
"""
|
||||
SELECT seq, role, content
|
||||
FROM messages
|
||||
WHERE session_id = ?
|
||||
ORDER BY seq DESC
|
||||
""",
|
||||
(session_id,),
|
||||
).fetchall()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
if not rows:
|
||||
return []
|
||||
|
||||
# Walk newest-to-oldest counting *visible* user turns (actual user text,
|
||||
# not tool_result injections). Record the seq of every visible user
|
||||
# message so we can find a clean cut point later.
|
||||
visible_turn_seqs: List[int] = [] # newest first
|
||||
for seq, role, raw_content in rows:
|
||||
if role != "user":
|
||||
continue
|
||||
try:
|
||||
content = json.loads(raw_content)
|
||||
except Exception:
|
||||
content = raw_content
|
||||
if _is_visible_user_message(content):
|
||||
visible_turn_seqs.append(seq)
|
||||
|
||||
# Determine the seq of the oldest visible user message we want to keep.
|
||||
# If the total turns fit within max_turns, keep everything.
|
||||
if len(visible_turn_seqs) <= max_turns:
|
||||
cutoff_seq = None # keep all
|
||||
else:
|
||||
# The Nth visible user message (0-indexed) is the oldest we keep.
|
||||
cutoff_seq = visible_turn_seqs[max_turns - 1]
|
||||
|
||||
# Build result in chronological order, starting from cutoff.
|
||||
# IMPORTANT: we start exactly at cutoff_seq (the visible user message),
|
||||
# never mid-group, so tool_use / tool_result pairs are always complete.
|
||||
result = []
|
||||
for seq, role, raw_content in reversed(rows):
|
||||
if cutoff_seq is not None and seq < cutoff_seq:
|
||||
continue
|
||||
try:
|
||||
content = json.loads(raw_content)
|
||||
except Exception:
|
||||
content = raw_content
|
||||
result.append({"role": role, "content": content})
|
||||
return result
|
||||
|
||||
def append_messages(
|
||||
self,
|
||||
session_id: str,
|
||||
messages: List[Dict[str, Any]],
|
||||
channel_type: str = "",
|
||||
) -> None:
|
||||
"""
|
||||
Append new messages to a session's history.
|
||||
|
||||
Seq numbers continue from the session's current maximum, so
|
||||
concurrent callers on distinct sessions never collide.
|
||||
|
||||
Args:
|
||||
session_id: Unique session identifier.
|
||||
messages: List of message dicts to append.
|
||||
channel_type: Source channel (e.g. "feishu", "web", "wechat").
|
||||
Only written on session creation; ignored on update.
|
||||
"""
|
||||
if not messages:
|
||||
return
|
||||
|
||||
now = int(time.time())
|
||||
with self._lock:
|
||||
conn = self._connect()
|
||||
try:
|
||||
with conn:
|
||||
# INSERT OR IGNORE creates the row on first visit;
|
||||
# the UPDATE always refreshes last_active.
|
||||
# Avoids ON CONFLICT...DO UPDATE (requires SQLite >= 3.24).
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT OR IGNORE INTO sessions
|
||||
(session_id, channel_type, created_at, last_active, msg_count)
|
||||
VALUES (?, ?, ?, ?, 0)
|
||||
""",
|
||||
(session_id, channel_type, now, now),
|
||||
)
|
||||
conn.execute(
|
||||
"UPDATE sessions SET last_active = ? WHERE session_id = ?",
|
||||
(now, session_id),
|
||||
)
|
||||
|
||||
# Determine starting seq for the new batch.
|
||||
row = conn.execute(
|
||||
"SELECT COALESCE(MAX(seq), -1) FROM messages WHERE session_id = ?",
|
||||
(session_id,),
|
||||
).fetchone()
|
||||
next_seq = row[0] + 1
|
||||
|
||||
for msg in messages:
|
||||
role = msg.get("role", "")
|
||||
content = json.dumps(
|
||||
msg.get("content", ""), ensure_ascii=False
|
||||
)
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT OR IGNORE INTO messages
|
||||
(session_id, seq, role, content, created_at)
|
||||
VALUES (?, ?, ?, ?, ?)
|
||||
""",
|
||||
(session_id, next_seq, role, content, now),
|
||||
)
|
||||
next_seq += 1
|
||||
|
||||
conn.execute(
|
||||
"""
|
||||
UPDATE sessions
|
||||
SET msg_count = (
|
||||
SELECT COUNT(*) FROM messages WHERE session_id = ?
|
||||
)
|
||||
WHERE session_id = ?
|
||||
""",
|
||||
(session_id, session_id),
|
||||
)
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def clear_session(self, session_id: str) -> None:
|
||||
"""Delete all messages and the session record for a given session_id."""
|
||||
with self._lock:
|
||||
conn = self._connect()
|
||||
try:
|
||||
with conn:
|
||||
conn.execute(
|
||||
"DELETE FROM messages WHERE session_id = ?", (session_id,)
|
||||
)
|
||||
conn.execute(
|
||||
"DELETE FROM sessions WHERE session_id = ?", (session_id,)
|
||||
)
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def cleanup_old_sessions(self, max_age_days: Optional[int] = None) -> int:
|
||||
"""
|
||||
Delete sessions that have not been active within max_age_days.
|
||||
|
||||
Args:
|
||||
max_age_days: Override the default retention period.
|
||||
|
||||
Returns:
|
||||
Number of sessions deleted.
|
||||
"""
|
||||
try:
|
||||
from config import conf
|
||||
max_age = max_age_days or conf().get(
|
||||
"conversation_max_age_days", DEFAULT_MAX_AGE_DAYS
|
||||
)
|
||||
except Exception:
|
||||
max_age = max_age_days or DEFAULT_MAX_AGE_DAYS
|
||||
|
||||
cutoff = int(time.time()) - max_age * 86400
|
||||
deleted = 0
|
||||
|
||||
with self._lock:
|
||||
conn = self._connect()
|
||||
try:
|
||||
with conn:
|
||||
stale = conn.execute(
|
||||
"SELECT session_id FROM sessions WHERE last_active < ?",
|
||||
(cutoff,),
|
||||
).fetchall()
|
||||
for (sid,) in stale:
|
||||
conn.execute(
|
||||
"DELETE FROM messages WHERE session_id = ?", (sid,)
|
||||
)
|
||||
conn.execute(
|
||||
"DELETE FROM sessions WHERE session_id = ?", (sid,)
|
||||
)
|
||||
deleted += 1
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
if deleted:
|
||||
logger.info(f"[ConversationStore] Pruned {deleted} expired sessions")
|
||||
return deleted
|
||||
|
||||
def load_history_page(
|
||||
self,
|
||||
session_id: str,
|
||||
page: int = 1,
|
||||
page_size: int = 20,
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Load a page of conversation history for UI display, grouped into turns.
|
||||
|
||||
Each "turn" maps to one of:
|
||||
- A user message (role="user", content=str)
|
||||
- An assistant message (role="assistant", content=str,
|
||||
tool_calls=[{name, arguments, result}] when tools were used)
|
||||
|
||||
Internal tool_result user messages are merged into the preceding
|
||||
assistant entry's tool_calls list and never appear as standalone items.
|
||||
|
||||
Pages are numbered from 1 (most recent). Messages within a page are
|
||||
returned in chronological order.
|
||||
|
||||
Returns:
|
||||
{
|
||||
"messages": [
|
||||
{
|
||||
"role": "user" | "assistant",
|
||||
"content": str,
|
||||
"tool_calls": [...], # assistant only, may be []
|
||||
"created_at": int,
|
||||
},
|
||||
...
|
||||
],
|
||||
"total": <visible turn count>,
|
||||
"page": <current page>,
|
||||
"page_size": <page_size>,
|
||||
"has_more": bool,
|
||||
}
|
||||
"""
|
||||
page = max(1, page)
|
||||
with self._lock:
|
||||
conn = self._connect()
|
||||
try:
|
||||
rows = conn.execute(
|
||||
"""
|
||||
SELECT role, content, created_at
|
||||
FROM messages
|
||||
WHERE session_id = ?
|
||||
ORDER BY seq ASC
|
||||
""",
|
||||
(session_id,),
|
||||
).fetchall()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
visible = _group_into_display_turns(rows)
|
||||
|
||||
total = len(visible)
|
||||
offset = (page - 1) * page_size
|
||||
page_items = list(reversed(visible))[offset: offset + page_size]
|
||||
page_items = list(reversed(page_items))
|
||||
|
||||
return {
|
||||
"messages": page_items,
|
||||
"total": total,
|
||||
"page": page,
|
||||
"page_size": page_size,
|
||||
"has_more": offset + page_size < total,
|
||||
}
|
||||
|
||||
def get_stats(self) -> Dict[str, Any]:
|
||||
"""Return basic stats keyed by channel_type, for monitoring."""
|
||||
with self._lock:
|
||||
conn = self._connect()
|
||||
try:
|
||||
total_sessions = conn.execute(
|
||||
"SELECT COUNT(*) FROM sessions"
|
||||
).fetchone()[0]
|
||||
total_messages = conn.execute(
|
||||
"SELECT COUNT(*) FROM messages"
|
||||
).fetchone()[0]
|
||||
by_channel = conn.execute(
|
||||
"""
|
||||
SELECT channel_type, COUNT(*) as cnt
|
||||
FROM sessions
|
||||
GROUP BY channel_type
|
||||
ORDER BY cnt DESC
|
||||
"""
|
||||
).fetchall()
|
||||
return {
|
||||
"total_sessions": total_sessions,
|
||||
"total_messages": total_messages,
|
||||
"by_channel": {row[0] or "unknown": row[1] for row in by_channel},
|
||||
}
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Internal helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _init_db(self) -> None:
|
||||
self._db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
conn = self._connect()
|
||||
try:
|
||||
conn.executescript(_DDL)
|
||||
conn.commit()
|
||||
self._migrate(conn)
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def _migrate(self, conn: sqlite3.Connection) -> None:
|
||||
"""Apply incremental schema migrations on existing databases."""
|
||||
cols = {
|
||||
row[1]
|
||||
for row in conn.execute("PRAGMA table_info(sessions)").fetchall()
|
||||
}
|
||||
if "channel_type" not in cols:
|
||||
try:
|
||||
conn.execute(_MIGRATION_ADD_CHANNEL_TYPE)
|
||||
conn.commit()
|
||||
logger.info("[ConversationStore] Migrated: added channel_type column")
|
||||
except Exception as e:
|
||||
logger.warning(f"[ConversationStore] Migration failed: {e}")
|
||||
|
||||
def _connect(self) -> sqlite3.Connection:
|
||||
conn = sqlite3.connect(str(self._db_path), timeout=10)
|
||||
conn.execute("PRAGMA journal_mode=WAL")
|
||||
conn.execute("PRAGMA synchronous=NORMAL")
|
||||
return conn
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Singleton
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_store_instance: Optional[ConversationStore] = None
|
||||
_store_lock = threading.Lock()
|
||||
|
||||
|
||||
def get_conversation_store() -> ConversationStore:
|
||||
"""
|
||||
Return the process-wide ConversationStore singleton.
|
||||
|
||||
Reuses the long-term memory database so the project stays with a single
|
||||
SQLite file: ~/cow/memory/long-term/index.db
|
||||
The conversation tables (sessions / messages) are separate from the
|
||||
memory tables (memory_chunks / file_metadata) — no conflicts.
|
||||
"""
|
||||
global _store_instance
|
||||
if _store_instance is not None:
|
||||
return _store_instance
|
||||
|
||||
with _store_lock:
|
||||
if _store_instance is not None:
|
||||
return _store_instance
|
||||
|
||||
try:
|
||||
from agent.memory.config import get_default_memory_config
|
||||
db_path = get_default_memory_config().get_db_path()
|
||||
except Exception:
|
||||
from common.utils import expand_path
|
||||
db_path = Path(expand_path("~/cow")) / "memory" / "long-term" / "index.db"
|
||||
|
||||
_store_instance = ConversationStore(db_path)
|
||||
logger.debug(f"[ConversationStore] Using shared DB at: {db_path}")
|
||||
return _store_instance
|
||||
@@ -501,7 +501,7 @@ class AgentStreamExecutor:
|
||||
|
||||
# Prepare messages
|
||||
messages = self._prepare_messages()
|
||||
logger.debug(f"Sending {len(messages)} messages to LLM")
|
||||
logger.info(f"Sending {len(messages)} messages to LLM")
|
||||
|
||||
# Prepare tool definitions (OpenAI/Claude format)
|
||||
tools_schema = None
|
||||
@@ -574,7 +574,7 @@ class AgentStreamExecutor:
|
||||
raise Exception(f"{error_msg} (Status: {status_code}, Code: {error_code}, Type: {error_type})")
|
||||
|
||||
# Parse chunk
|
||||
if isinstance(chunk, dict) and "choices" in chunk:
|
||||
if isinstance(chunk, dict) and chunk.get("choices"):
|
||||
choice = chunk["choices"][0]
|
||||
delta = choice.get("delta", {})
|
||||
|
||||
|
||||
159
app.py
159
app.py
@@ -13,7 +13,6 @@ from plugins import *
|
||||
import threading
|
||||
|
||||
|
||||
# Global channel manager for restart support
|
||||
_channel_mgr = None
|
||||
|
||||
|
||||
@@ -21,92 +20,130 @@ def get_channel_manager():
|
||||
return _channel_mgr
|
||||
|
||||
|
||||
def _parse_channel_type(raw) -> list:
|
||||
"""
|
||||
Parse channel_type config value into a list of channel names.
|
||||
Supports:
|
||||
- single string: "feishu"
|
||||
- comma-separated string: "feishu, dingtalk"
|
||||
- list: ["feishu", "dingtalk"]
|
||||
"""
|
||||
if isinstance(raw, list):
|
||||
return [ch.strip() for ch in raw if ch.strip()]
|
||||
if isinstance(raw, str):
|
||||
return [ch.strip() for ch in raw.split(",") if ch.strip()]
|
||||
return []
|
||||
|
||||
|
||||
class ChannelManager:
|
||||
"""
|
||||
Manage the lifecycle of a channel, supporting restart from sub-threads.
|
||||
The channel.startup() runs in a daemon thread so that the main thread
|
||||
remains available and a new channel can be started at any time.
|
||||
Manage the lifecycle of multiple channels running concurrently.
|
||||
Each channel.startup() runs in its own daemon thread.
|
||||
The web channel is started as default console unless explicitly disabled.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self._channel = None
|
||||
self._channel_thread = None
|
||||
self._channels = {} # channel_name -> channel instance
|
||||
self._threads = {} # channel_name -> thread
|
||||
self._primary_channel = None
|
||||
self._lock = threading.Lock()
|
||||
|
||||
@property
|
||||
def channel(self):
|
||||
return self._channel
|
||||
"""Return the primary (first non-web) channel for backward compatibility."""
|
||||
return self._primary_channel
|
||||
|
||||
def start(self, channel_name: str, first_start: bool = False):
|
||||
def get_channel(self, channel_name: str):
|
||||
return self._channels.get(channel_name)
|
||||
|
||||
def start(self, channel_names: list, first_start: bool = False):
|
||||
"""
|
||||
Create and start a channel in a sub-thread.
|
||||
Create and start one or more channels in sub-threads.
|
||||
If first_start is True, plugins and linkai client will also be initialized.
|
||||
"""
|
||||
with self._lock:
|
||||
channel = channel_factory.create_channel(channel_name)
|
||||
self._channel = channel
|
||||
channels = []
|
||||
for name in channel_names:
|
||||
ch = channel_factory.create_channel(name)
|
||||
self._channels[name] = ch
|
||||
channels.append((name, ch))
|
||||
if self._primary_channel is None and name != "web":
|
||||
self._primary_channel = ch
|
||||
|
||||
if self._primary_channel is None and channels:
|
||||
self._primary_channel = channels[0][1]
|
||||
|
||||
if first_start:
|
||||
if channel_name in ["wx", "wxy", "terminal", "wechatmp", "web",
|
||||
"wechatmp_service", "wechatcom_app", "wework",
|
||||
const.FEISHU, const.DINGTALK]:
|
||||
PluginManager().load_plugins()
|
||||
PluginManager().load_plugins()
|
||||
|
||||
if conf().get("use_linkai"):
|
||||
try:
|
||||
from common import cloud_client
|
||||
threading.Thread(target=cloud_client.start, args=(channel, self), daemon=True).start()
|
||||
except Exception as e:
|
||||
threading.Thread(
|
||||
target=cloud_client.start,
|
||||
args=(self._primary_channel, self),
|
||||
daemon=True,
|
||||
).start()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Run channel.startup() in a daemon thread so we can restart later
|
||||
self._channel_thread = threading.Thread(
|
||||
target=self._run_channel, args=(channel,), daemon=True
|
||||
)
|
||||
self._channel_thread.start()
|
||||
logger.debug(f"[ChannelManager] Channel '{channel_name}' started in sub-thread")
|
||||
# Start web console first so its logs print cleanly,
|
||||
# then start remaining channels after a brief pause.
|
||||
web_entry = None
|
||||
other_entries = []
|
||||
for entry in channels:
|
||||
if entry[0] == "web":
|
||||
web_entry = entry
|
||||
else:
|
||||
other_entries.append(entry)
|
||||
|
||||
def _run_channel(self, channel):
|
||||
ordered = ([web_entry] if web_entry else []) + other_entries
|
||||
for i, (name, ch) in enumerate(ordered):
|
||||
if i > 0 and name != "web":
|
||||
time.sleep(0.1)
|
||||
t = threading.Thread(target=self._run_channel, args=(name, ch), daemon=True)
|
||||
self._threads[name] = t
|
||||
t.start()
|
||||
logger.debug(f"[ChannelManager] Channel '{name}' started in sub-thread")
|
||||
|
||||
def _run_channel(self, name: str, channel):
|
||||
try:
|
||||
channel.startup()
|
||||
except Exception as e:
|
||||
logger.error(f"[ChannelManager] Channel startup error: {e}")
|
||||
logger.error(f"[ChannelManager] Channel '{name}' startup error: {e}")
|
||||
logger.exception(e)
|
||||
|
||||
def stop(self):
|
||||
def stop(self, channel_name: str = None):
|
||||
"""
|
||||
Stop the current channel. Since most channel startup() methods block
|
||||
on an HTTP server or stream client, we stop by terminating the thread.
|
||||
Stop channel(s). If channel_name is given, stop only that channel;
|
||||
otherwise stop all channels.
|
||||
"""
|
||||
with self._lock:
|
||||
if self._channel is None:
|
||||
return
|
||||
channel_type = getattr(self._channel, 'channel_type', 'unknown')
|
||||
logger.info(f"[ChannelManager] Stopping channel '{channel_type}'...")
|
||||
|
||||
# Try graceful stop if channel implements it
|
||||
try:
|
||||
if hasattr(self._channel, 'stop'):
|
||||
self._channel.stop()
|
||||
except Exception as e:
|
||||
logger.warning(f"[ChannelManager] Error during channel stop: {e}")
|
||||
|
||||
self._channel = None
|
||||
self._channel_thread = None
|
||||
names = [channel_name] if channel_name else list(self._channels.keys())
|
||||
for name in names:
|
||||
ch = self._channels.pop(name, None)
|
||||
self._threads.pop(name, None)
|
||||
if ch is None:
|
||||
continue
|
||||
logger.info(f"[ChannelManager] Stopping channel '{name}'...")
|
||||
try:
|
||||
if hasattr(ch, 'stop'):
|
||||
ch.stop()
|
||||
except Exception as e:
|
||||
logger.warning(f"[ChannelManager] Error during channel '{name}' stop: {e}")
|
||||
if channel_name and self._primary_channel is self._channels.get(channel_name):
|
||||
self._primary_channel = None
|
||||
|
||||
def restart(self, new_channel_name: str):
|
||||
"""
|
||||
Restart the channel with a new channel type.
|
||||
Restart a single channel with a new channel type.
|
||||
Can be called from any thread (e.g. linkai config callback).
|
||||
"""
|
||||
logger.info(f"[ChannelManager] Restarting channel to '{new_channel_name}'...")
|
||||
self.stop()
|
||||
|
||||
# Clear singleton cache so a fresh channel instance is created
|
||||
self.stop(new_channel_name)
|
||||
_clear_singleton_cache(new_channel_name)
|
||||
|
||||
time.sleep(1) # Brief pause to allow resources to release
|
||||
self.start(new_channel_name, first_start=False)
|
||||
time.sleep(1)
|
||||
self.start([new_channel_name], first_start=False)
|
||||
logger.info(f"[ChannelManager] Channel restarted to '{new_channel_name}' successfully")
|
||||
|
||||
|
||||
@@ -130,14 +167,11 @@ def _clear_singleton_cache(channel_name: str):
|
||||
module_path = cls_map.get(channel_name)
|
||||
if not module_path:
|
||||
return
|
||||
# The singleton decorator stores instances in a closure dict keyed by class.
|
||||
# We need to find the actual class and clear it from the closure.
|
||||
try:
|
||||
parts = module_path.rsplit(".", 1)
|
||||
module_name, class_name = parts[0], parts[1]
|
||||
import importlib
|
||||
module = importlib.import_module(module_name)
|
||||
# The module-level name is the wrapper function from @singleton
|
||||
wrapper = getattr(module, class_name, None)
|
||||
if wrapper and hasattr(wrapper, '__closure__') and wrapper.__closure__:
|
||||
for cell in wrapper.__closure__:
|
||||
@@ -176,17 +210,28 @@ def run():
|
||||
# kill signal
|
||||
sigterm_handler_wrap(signal.SIGTERM)
|
||||
|
||||
# create channel
|
||||
channel_name = conf().get("channel_type", "wx")
|
||||
# Parse channel_type into a list
|
||||
raw_channel = conf().get("channel_type", "wx")
|
||||
|
||||
if "--cmd" in sys.argv:
|
||||
channel_name = "terminal"
|
||||
channel_names = ["terminal"]
|
||||
else:
|
||||
channel_names = _parse_channel_type(raw_channel)
|
||||
if not channel_names:
|
||||
channel_names = ["wx"]
|
||||
|
||||
if channel_name == "wxy":
|
||||
if "wxy" in channel_names:
|
||||
os.environ["WECHATY_LOG"] = "warn"
|
||||
|
||||
# Auto-start web console unless explicitly disabled
|
||||
web_console_enabled = conf().get("web_console", True)
|
||||
if web_console_enabled and "web" not in channel_names:
|
||||
channel_names.append("web")
|
||||
|
||||
logger.info(f"[App] Starting channels: {channel_names}")
|
||||
|
||||
_channel_mgr = ChannelManager()
|
||||
_channel_mgr.start(channel_name, first_start=True)
|
||||
_channel_mgr.start(channel_names, first_start=True)
|
||||
|
||||
while True:
|
||||
time.sleep(1)
|
||||
|
||||
@@ -325,6 +325,10 @@ class AgentBridge:
|
||||
logger.warning(f"[AgentBridge] Failed to attach context to scheduler: {e}")
|
||||
break
|
||||
|
||||
# Record message count before execution so we can diff new messages
|
||||
with agent.messages_lock:
|
||||
pre_run_len = len(agent.messages)
|
||||
|
||||
try:
|
||||
# Use agent's run_stream method with event handler
|
||||
response = agent.run_stream(
|
||||
@@ -336,9 +340,16 @@ class AgentBridge:
|
||||
# Restore original tools
|
||||
if context and context.get("is_scheduled_task"):
|
||||
agent.tools = original_tools
|
||||
|
||||
|
||||
# Log execution summary
|
||||
event_handler.log_summary()
|
||||
|
||||
# Persist new messages generated during this run
|
||||
if session_id:
|
||||
channel_type = (context.get("channel_type") or "") if context else ""
|
||||
with agent.messages_lock:
|
||||
new_messages = agent.messages[pre_run_len:]
|
||||
self._persist_messages(session_id, list(new_messages), channel_type)
|
||||
|
||||
# Check if there are files to send (from read tool)
|
||||
if hasattr(agent, 'stream_executor') and hasattr(agent.stream_executor, 'files_to_send'):
|
||||
@@ -475,6 +486,32 @@ class AgentBridge:
|
||||
except Exception as e:
|
||||
logger.warning(f"[AgentBridge] Failed to migrate API keys: {e}")
|
||||
|
||||
def _persist_messages(
|
||||
self, session_id: str, new_messages: list, channel_type: str = ""
|
||||
) -> None:
|
||||
"""
|
||||
Persist new messages to the conversation store after each agent run.
|
||||
|
||||
Failures are logged but never propagate — they must not interrupt replies.
|
||||
"""
|
||||
if not new_messages:
|
||||
return
|
||||
try:
|
||||
from config import conf
|
||||
if not conf().get("conversation_persistence", True):
|
||||
return
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
from agent.memory import get_conversation_store
|
||||
get_conversation_store().append_messages(
|
||||
session_id, new_messages, channel_type=channel_type
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"[AgentBridge] Failed to persist messages for session={session_id}: {e}"
|
||||
)
|
||||
|
||||
def clear_session(self, session_id: str):
|
||||
"""
|
||||
Clear a specific session's agent and conversation history
|
||||
|
||||
@@ -118,8 +118,47 @@ class AgentInitializer:
|
||||
# Attach memory manager
|
||||
if memory_manager:
|
||||
agent.memory_manager = memory_manager
|
||||
|
||||
|
||||
# Restore persisted conversation history for this session
|
||||
if session_id:
|
||||
self._restore_conversation_history(agent, session_id)
|
||||
|
||||
return agent
|
||||
|
||||
def _restore_conversation_history(self, agent, session_id: str) -> None:
|
||||
"""
|
||||
Load persisted conversation messages from SQLite and inject them
|
||||
into the agent's in-memory message list.
|
||||
|
||||
Only runs when conversation persistence is enabled (default: True).
|
||||
Respects agent_max_context_turns to limit how many turns are loaded.
|
||||
"""
|
||||
from config import conf
|
||||
if not conf().get("conversation_persistence", True):
|
||||
return
|
||||
|
||||
try:
|
||||
from agent.memory import get_conversation_store
|
||||
store = get_conversation_store()
|
||||
# On restore, load at most min(10, max_turns // 2) turns so that
|
||||
# a long-running session does not immediately fill the context window
|
||||
# after a restart. The full max_turns budget is reserved for the
|
||||
# live conversation that follows.
|
||||
max_turns = conf().get("agent_max_context_turns", 30)
|
||||
restore_turns = min(6, max(1, max_turns // 3))
|
||||
saved = store.load_messages(session_id, max_turns=restore_turns)
|
||||
if saved:
|
||||
with agent.messages_lock:
|
||||
agent.messages = saved
|
||||
logger.debug(
|
||||
f"[AgentInitializer] Restored {len(saved)} messages "
|
||||
f"({restore_turns} turns cap) for session={session_id}"
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"[AgentInitializer] Failed to restore conversation history for "
|
||||
f"session={session_id}: {e}"
|
||||
)
|
||||
|
||||
def _load_env_file(self):
|
||||
"""Load environment variables from .env file"""
|
||||
|
||||
@@ -24,11 +24,16 @@ handler_pool = ThreadPoolExecutor(max_workers=8) # 处理消息的线程池
|
||||
class ChatChannel(Channel):
|
||||
name = None # 登录的用户名
|
||||
user_id = None # 登录的用户id
|
||||
futures = {} # 记录每个session_id提交到线程池的future对象, 用于重置会话时把没执行的future取消掉,正在执行的不会被取消
|
||||
sessions = {} # 用于控制并发,每个session_id同时只能有一个context在处理
|
||||
lock = threading.Lock() # 用于控制对sessions的访问
|
||||
|
||||
def __init__(self):
|
||||
# Instance-level attributes so each channel subclass has its own
|
||||
# independent session queue and lock. Previously these were class-level,
|
||||
# which caused contexts from one channel (e.g. Feishu) to be consumed
|
||||
# by another channel's consume() thread (e.g. Web), leading to errors
|
||||
# like "No request_id found in context".
|
||||
self.futures = {}
|
||||
self.sessions = {}
|
||||
self.lock = threading.Lock()
|
||||
_thread = threading.Thread(target=self.consume)
|
||||
_thread.setDaemon(True)
|
||||
_thread.start()
|
||||
|
||||
@@ -43,8 +43,17 @@
|
||||
}
|
||||
</script>
|
||||
<link rel="stylesheet" href="assets/css/console.css">
|
||||
<!-- Apply theme/lang before first paint to avoid flash of unstyled content.
|
||||
This runs synchronously in <head> so the correct class is on <html>
|
||||
before any CSS or body rendering occurs. -->
|
||||
<script>
|
||||
(function() {
|
||||
var theme = localStorage.getItem('cow_theme') || 'dark';
|
||||
if (theme === 'dark') document.documentElement.classList.add('dark');
|
||||
})();
|
||||
</script>
|
||||
</head>
|
||||
<body class="h-screen overflow-hidden bg-gray-50 dark:bg-[#111111] text-slate-800 dark:text-slate-200 font-sans transition-colors duration-200">
|
||||
<body class="h-screen overflow-hidden bg-gray-50 dark:bg-[#111111] text-slate-800 dark:text-slate-200 font-sans">
|
||||
<div id="app" class="flex h-screen">
|
||||
|
||||
<!-- ================================================================ -->
|
||||
|
||||
@@ -232,19 +232,37 @@ function renderMarkdown(text) {
|
||||
// =====================================================================
|
||||
// Chat Module
|
||||
// =====================================================================
|
||||
let sessionId = generateSessionId();
|
||||
let isPolling = false;
|
||||
let loadingContainers = {};
|
||||
let activeStreams = {}; // request_id -> EventSource
|
||||
let isComposing = false;
|
||||
let appConfig = { use_agent: false, title: 'CowAgent', subtitle: '' };
|
||||
|
||||
const SESSION_ID_KEY = 'cow_session_id';
|
||||
|
||||
function generateSessionId() {
|
||||
return 'session_' + ([1e7]+-1e3+-4e3+-8e3+-1e11).replace(/[018]/g, c =>
|
||||
(c ^ crypto.getRandomValues(new Uint8Array(1))[0] & 15 >> c / 4).toString(16)
|
||||
);
|
||||
}
|
||||
|
||||
// Restore session_id from localStorage so conversation history survives page refresh.
|
||||
// A new id is only generated when the user explicitly starts a new chat.
|
||||
function loadOrCreateSessionId() {
|
||||
const stored = localStorage.getItem(SESSION_ID_KEY);
|
||||
if (stored) return stored;
|
||||
const fresh = generateSessionId();
|
||||
localStorage.setItem(SESSION_ID_KEY, fresh);
|
||||
return fresh;
|
||||
}
|
||||
|
||||
let sessionId = loadOrCreateSessionId();
|
||||
|
||||
// ---- Conversation history state ----
|
||||
let historyPage = 0; // last page fetched (0 = nothing fetched yet)
|
||||
let historyHasMore = false;
|
||||
let historyLoading = false;
|
||||
|
||||
fetch('/config').then(r => r.json()).then(data => {
|
||||
if (data.status === 'success') {
|
||||
appConfig = data;
|
||||
@@ -257,14 +275,20 @@ fetch('/config').then(r => r.json()).then(data => {
|
||||
document.getElementById('cfg-max-steps').textContent = data.agent_max_steps || '--';
|
||||
document.getElementById('cfg-channel').textContent = data.channel_type || '--';
|
||||
}
|
||||
}).catch(() => {});
|
||||
// Load conversation history after config is ready
|
||||
loadHistory(1);
|
||||
}).catch(() => { loadHistory(1); });
|
||||
|
||||
const chatInput = document.getElementById('chat-input');
|
||||
const sendBtn = document.getElementById('send-btn');
|
||||
const messagesDiv = document.getElementById('chat-messages');
|
||||
|
||||
chatInput.addEventListener('compositionstart', () => { isComposing = true; });
|
||||
chatInput.addEventListener('compositionend', () => { isComposing = false; });
|
||||
// Safari fires compositionend *before* the confirming keydown event, so if we
|
||||
// reset isComposing synchronously the keydown handler sees !isComposing and
|
||||
// sends the message prematurely. A setTimeout(0) defers the reset until after
|
||||
// keydown has been processed, fixing the Safari IME Enter-to-confirm bug.
|
||||
chatInput.addEventListener('compositionend', () => { setTimeout(() => { isComposing = false; }, 0); });
|
||||
|
||||
chatInput.addEventListener('input', function() {
|
||||
this.style.height = '42px';
|
||||
@@ -530,7 +554,7 @@ function startPolling() {
|
||||
poll();
|
||||
}
|
||||
|
||||
function addUserMessage(content, timestamp) {
|
||||
function createUserMessageEl(content, timestamp) {
|
||||
const el = document.createElement('div');
|
||||
el.className = 'flex justify-end px-4 sm:px-6 py-3';
|
||||
el.innerHTML = `
|
||||
@@ -541,28 +565,141 @@ function addUserMessage(content, timestamp) {
|
||||
<div class="text-xs text-slate-400 dark:text-slate-500 mt-1.5 text-right">${formatTime(timestamp)}</div>
|
||||
</div>
|
||||
`;
|
||||
return el;
|
||||
}
|
||||
|
||||
function renderToolCallsHtml(toolCalls) {
|
||||
if (!toolCalls || toolCalls.length === 0) return '';
|
||||
return toolCalls.map(tc => {
|
||||
const argsStr = formatToolArgs(tc.arguments || {});
|
||||
const resultStr = tc.result ? escapeHtml(String(tc.result)) : '';
|
||||
const hasResult = !!resultStr;
|
||||
return `
|
||||
<div class="agent-step agent-tool-step">
|
||||
<div class="tool-header" onclick="this.parentElement.classList.toggle('expanded')">
|
||||
<i class="fas fa-check text-primary-400 flex-shrink-0 tool-icon"></i>
|
||||
<span class="tool-name">${escapeHtml(tc.name || '')}</span>
|
||||
<i class="fas fa-chevron-right tool-chevron"></i>
|
||||
</div>
|
||||
<div class="tool-detail">
|
||||
<div class="tool-detail-section">
|
||||
<div class="tool-detail-label">Input</div>
|
||||
<pre class="tool-detail-content">${argsStr}</pre>
|
||||
</div>
|
||||
${hasResult ? `
|
||||
<div class="tool-detail-section tool-output-section">
|
||||
<div class="tool-detail-label">Output</div>
|
||||
<pre class="tool-detail-content">${resultStr}</pre>
|
||||
</div>` : ''}
|
||||
</div>
|
||||
</div>`;
|
||||
}).join('');
|
||||
}
|
||||
|
||||
function createBotMessageEl(content, timestamp, requestId, toolCalls) {
|
||||
const el = document.createElement('div');
|
||||
el.className = 'flex gap-3 px-4 sm:px-6 py-3';
|
||||
if (requestId) el.dataset.requestId = requestId;
|
||||
const toolsHtml = renderToolCallsHtml(toolCalls);
|
||||
el.innerHTML = `
|
||||
<img src="assets/logo.jpg" alt="CowAgent" class="w-8 h-8 rounded-lg flex-shrink-0">
|
||||
<div class="min-w-0 flex-1 max-w-[85%]">
|
||||
<div class="bg-white dark:bg-[#1A1A1A] border border-slate-200 dark:border-white/10 rounded-2xl px-4 py-3 text-sm leading-relaxed msg-content text-slate-700 dark:text-slate-200">
|
||||
${toolsHtml ? `<div class="agent-steps">${toolsHtml}</div>` : ''}
|
||||
<div class="answer-content">${renderMarkdown(content)}</div>
|
||||
</div>
|
||||
<div class="text-xs text-slate-400 dark:text-slate-500 mt-1.5">${formatTime(timestamp)}</div>
|
||||
</div>
|
||||
`;
|
||||
applyHighlighting(el);
|
||||
return el;
|
||||
}
|
||||
|
||||
function addUserMessage(content, timestamp) {
|
||||
const el = createUserMessageEl(content, timestamp);
|
||||
messagesDiv.appendChild(el);
|
||||
scrollChatToBottom();
|
||||
}
|
||||
|
||||
function addBotMessage(content, timestamp, requestId) {
|
||||
const el = document.createElement('div');
|
||||
el.className = 'flex gap-3 px-4 sm:px-6 py-3';
|
||||
if (requestId) el.dataset.requestId = requestId;
|
||||
el.innerHTML = `
|
||||
<img src="assets/logo.jpg" alt="CowAgent" class="w-8 h-8 rounded-lg flex-shrink-0">
|
||||
<div class="min-w-0 flex-1 max-w-[85%]">
|
||||
<div class="bg-white dark:bg-[#1A1A1A] border border-slate-200 dark:border-white/10 rounded-2xl px-4 py-3 text-sm leading-relaxed msg-content text-slate-700 dark:text-slate-200">
|
||||
${renderMarkdown(content)}
|
||||
</div>
|
||||
<div class="text-xs text-slate-400 dark:text-slate-500 mt-1.5">${formatTime(timestamp)}</div>
|
||||
</div>
|
||||
`;
|
||||
const el = createBotMessageEl(content, timestamp, requestId);
|
||||
messagesDiv.appendChild(el);
|
||||
applyHighlighting(el);
|
||||
scrollChatToBottom();
|
||||
}
|
||||
|
||||
// Load conversation history from the server (page 1 = most recent messages).
|
||||
// Subsequent pages prepend older messages when the user scrolls to the top.
|
||||
function loadHistory(page) {
|
||||
if (historyLoading) return;
|
||||
historyLoading = true;
|
||||
|
||||
fetch(`/api/history?session_id=${encodeURIComponent(sessionId)}&page=${page}&page_size=20`)
|
||||
.then(r => r.json())
|
||||
.then(data => {
|
||||
if (data.status !== 'success' || data.messages.length === 0) return;
|
||||
|
||||
const prevScrollHeight = messagesDiv.scrollHeight;
|
||||
const isFirstLoad = page === 1;
|
||||
|
||||
// On first load, remove the welcome screen if history exists
|
||||
if (isFirstLoad) {
|
||||
const ws = document.getElementById('welcome-screen');
|
||||
if (ws) ws.remove();
|
||||
}
|
||||
|
||||
// Build a fragment of history message elements in chronological order
|
||||
const fragment = document.createDocumentFragment();
|
||||
|
||||
if (data.has_more && page > 1) {
|
||||
// Keep the "load more" sentinel in place (inserted below)
|
||||
}
|
||||
|
||||
data.messages.forEach(msg => {
|
||||
const hasContent = msg.content && msg.content.trim();
|
||||
const hasToolCalls = msg.role === 'assistant' && msg.tool_calls && msg.tool_calls.length > 0;
|
||||
if (!hasContent && !hasToolCalls) return;
|
||||
const ts = new Date(msg.created_at * 1000);
|
||||
const el = msg.role === 'user'
|
||||
? createUserMessageEl(msg.content, ts)
|
||||
: createBotMessageEl(msg.content || '', ts, null, msg.tool_calls);
|
||||
fragment.appendChild(el);
|
||||
});
|
||||
|
||||
// Prepend history above any existing messages
|
||||
const sentinel = document.getElementById('history-load-more');
|
||||
const insertBefore = sentinel ? sentinel.nextSibling : messagesDiv.firstChild;
|
||||
messagesDiv.insertBefore(fragment, insertBefore);
|
||||
|
||||
// Manage the "load more" sentinel at the very top
|
||||
if (data.has_more) {
|
||||
if (!document.getElementById('history-load-more')) {
|
||||
const btn = document.createElement('div');
|
||||
btn.id = 'history-load-more';
|
||||
btn.className = 'flex justify-center py-3';
|
||||
btn.innerHTML = `<button class="text-xs text-slate-400 dark:text-slate-500 hover:text-primary-400 transition-colors" onclick="loadHistory(historyPage + 1)">Load earlier messages</button>`;
|
||||
messagesDiv.insertBefore(btn, messagesDiv.firstChild);
|
||||
}
|
||||
} else {
|
||||
const sentinel = document.getElementById('history-load-more');
|
||||
if (sentinel) sentinel.remove();
|
||||
}
|
||||
|
||||
historyHasMore = data.has_more;
|
||||
historyPage = page;
|
||||
|
||||
if (isFirstLoad) {
|
||||
// Use requestAnimationFrame to ensure the DOM has fully rendered
|
||||
// before scrolling, otherwise scrollHeight may not reflect new content.
|
||||
requestAnimationFrame(() => scrollChatToBottom());
|
||||
} else {
|
||||
// Restore scroll position so loading older messages doesn't jump the view
|
||||
messagesDiv.scrollTop = messagesDiv.scrollHeight - prevScrollHeight;
|
||||
}
|
||||
})
|
||||
.catch(() => {})
|
||||
.finally(() => { historyLoading = false; });
|
||||
}
|
||||
|
||||
function addLoadingIndicator() {
|
||||
const el = document.createElement('div');
|
||||
el.className = 'flex gap-3 px-4 sm:px-6 py-3';
|
||||
@@ -586,7 +723,9 @@ function newChat() {
|
||||
Object.values(activeStreams).forEach(es => { try { es.close(); } catch (_) {} });
|
||||
activeStreams = {};
|
||||
|
||||
// Generate a fresh session and persist it so the next page load also starts clean
|
||||
sessionId = generateSessionId();
|
||||
localStorage.setItem(SESSION_ID_KEY, sessionId);
|
||||
isPolling = false;
|
||||
loadingContainers = {};
|
||||
messagesDiv.innerHTML = '';
|
||||
@@ -969,3 +1108,11 @@ applyTheme();
|
||||
applyI18n();
|
||||
document.getElementById('sidebar-version').textContent = `CowAgent ${APP_VERSION}`;
|
||||
chatInput.focus();
|
||||
|
||||
// Re-enable color transition AFTER first paint so the theme applied in <head>
|
||||
// doesn't produce an animated flash on load. The class is missing from the
|
||||
// body initially; adding it here means transitions only fire on user-triggered
|
||||
// theme toggles, not on page load.
|
||||
requestAnimationFrame(() => {
|
||||
document.body.classList.add('transition-colors', 'duration-200');
|
||||
});
|
||||
|
||||
@@ -1,9 +1,15 @@
|
||||
import sys
|
||||
import time
|
||||
import web
|
||||
import json
|
||||
import logging
|
||||
import mimetypes
|
||||
import os
|
||||
import threading
|
||||
import time
|
||||
import uuid
|
||||
from queue import Queue, Empty
|
||||
|
||||
import web
|
||||
|
||||
from bridge.context import *
|
||||
from bridge.reply import Reply, ReplyType
|
||||
from channel.chat_channel import ChatChannel, check_prefix
|
||||
@@ -11,20 +17,17 @@ from channel.chat_message import ChatMessage
|
||||
from common.log import logger
|
||||
from common.singleton import singleton
|
||||
from config import conf
|
||||
import os
|
||||
import mimetypes
|
||||
import threading
|
||||
import logging
|
||||
|
||||
|
||||
class WebMessage(ChatMessage):
|
||||
def __init__(
|
||||
self,
|
||||
msg_id,
|
||||
content,
|
||||
ctype=ContextType.TEXT,
|
||||
from_user_id="User",
|
||||
to_user_id="Chatgpt",
|
||||
other_user_id="Chatgpt",
|
||||
self,
|
||||
msg_id,
|
||||
content,
|
||||
ctype=ContextType.TEXT,
|
||||
from_user_id="User",
|
||||
to_user_id="Chatgpt",
|
||||
other_user_id="Chatgpt",
|
||||
):
|
||||
self.msg_id = msg_id
|
||||
self.ctype = ctype
|
||||
@@ -38,7 +41,7 @@ class WebMessage(ChatMessage):
|
||||
class WebChannel(ChatChannel):
|
||||
NOT_SUPPORT_REPLYTYPE = [ReplyType.VOICE]
|
||||
_instance = None
|
||||
|
||||
|
||||
# def __new__(cls):
|
||||
# if cls._instance is None:
|
||||
# cls._instance = super(WebChannel, cls).__new__(cls)
|
||||
@@ -47,12 +50,11 @@ class WebChannel(ChatChannel):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.msg_id_counter = 0
|
||||
self.session_queues = {} # session_id -> Queue (fallback polling)
|
||||
self.request_to_session = {} # request_id -> session_id
|
||||
self.sse_queues = {} # request_id -> Queue (SSE streaming)
|
||||
self.session_queues = {} # session_id -> Queue (fallback polling)
|
||||
self.request_to_session = {} # request_id -> session_id
|
||||
self.sse_queues = {} # request_id -> Queue (SSE streaming)
|
||||
self._http_server = None
|
||||
|
||||
|
||||
def _generate_msg_id(self):
|
||||
"""生成唯一的消息ID"""
|
||||
self.msg_id_counter += 1
|
||||
@@ -111,6 +113,7 @@ class WebChannel(ChatChannel):
|
||||
|
||||
def _make_sse_callback(self, request_id: str):
|
||||
"""Build an on_event callback that pushes agent stream events into the SSE queue."""
|
||||
|
||||
def on_event(event: dict):
|
||||
if request_id not in self.sse_queues:
|
||||
return
|
||||
@@ -237,28 +240,28 @@ class WebChannel(ChatChannel):
|
||||
data = web.data()
|
||||
json_data = json.loads(data)
|
||||
session_id = json_data.get('session_id')
|
||||
|
||||
|
||||
if not session_id or session_id not in self.session_queues:
|
||||
return json.dumps({"status": "error", "message": "Invalid session ID"})
|
||||
|
||||
|
||||
# 尝试从队列获取响应,不等待
|
||||
try:
|
||||
# 使用peek而不是get,这样如果前端没有成功处理,下次还能获取到
|
||||
response = self.session_queues[session_id].get(block=False)
|
||||
|
||||
|
||||
# 返回响应,包含请求ID以区分不同请求
|
||||
return json.dumps({
|
||||
"status": "success",
|
||||
"status": "success",
|
||||
"has_content": True,
|
||||
"content": response["content"],
|
||||
"request_id": response["request_id"],
|
||||
"timestamp": response["timestamp"]
|
||||
})
|
||||
|
||||
|
||||
except Empty:
|
||||
# 没有新响应
|
||||
return json.dumps({"status": "success", "has_content": False})
|
||||
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error polling response: {e}")
|
||||
return json.dumps({"status": "error", "message": str(e)})
|
||||
@@ -271,9 +274,10 @@ class WebChannel(ChatChannel):
|
||||
|
||||
def startup(self):
|
||||
port = conf().get("web_port", 9899)
|
||||
|
||||
|
||||
# 打印可用渠道类型提示
|
||||
logger.info("[WebChannel] 当前channel为web,可修改 config.json 配置文件中的 channel_type 字段进行切换。全部可用类型为:")
|
||||
logger.info(
|
||||
"[WebChannel] 全部可用通道如下,可修改 config.json 配置文件中的 channel_type 字段进行切换,多个通道用逗号分隔:")
|
||||
logger.info("[WebChannel] 1. web - 网页")
|
||||
logger.info("[WebChannel] 2. terminal - 终端")
|
||||
logger.info("[WebChannel] 3. feishu - 飞书")
|
||||
@@ -281,16 +285,16 @@ class WebChannel(ChatChannel):
|
||||
logger.info("[WebChannel] 5. wechatcom_app - 企微自建应用")
|
||||
logger.info("[WebChannel] 6. wechatmp - 个人公众号")
|
||||
logger.info("[WebChannel] 7. wechatmp_service - 企业公众号")
|
||||
logger.info("[WebChannel] ✅ Web控制台已运行")
|
||||
logger.info(f"[WebChannel] 🌐 本地访问: http://localhost:{port}")
|
||||
logger.info(f"[WebChannel] 🌍 服务器访问: http://YOUR_IP:{port} (请将YOUR_IP替换为服务器IP)")
|
||||
logger.info("[WebChannel] ✅ Web对话网页已运行")
|
||||
|
||||
|
||||
# 确保静态文件目录存在
|
||||
static_dir = os.path.join(os.path.dirname(__file__), 'static')
|
||||
if not os.path.exists(static_dir):
|
||||
os.makedirs(static_dir)
|
||||
logger.debug(f"[WebChannel] Created static directory: {static_dir}")
|
||||
|
||||
|
||||
urls = (
|
||||
'/', 'RootHandler',
|
||||
'/message', 'MessageHandler',
|
||||
@@ -302,18 +306,19 @@ class WebChannel(ChatChannel):
|
||||
'/api/memory', 'MemoryHandler',
|
||||
'/api/memory/content', 'MemoryContentHandler',
|
||||
'/api/scheduler', 'SchedulerHandler',
|
||||
'/api/history', 'HistoryHandler',
|
||||
'/api/logs', 'LogsHandler',
|
||||
'/assets/(.*)', 'AssetsHandler',
|
||||
)
|
||||
app = web.application(urls, globals(), autoreload=False)
|
||||
|
||||
|
||||
# 完全禁用web.py的HTTP日志输出
|
||||
web.httpserver.LogMiddleware.log = lambda self, status, environ: None
|
||||
|
||||
|
||||
# 配置web.py的日志级别为ERROR
|
||||
logging.getLogger("web").setLevel(logging.ERROR)
|
||||
logging.getLogger("web.httpserver").setLevel(logging.ERROR)
|
||||
|
||||
|
||||
# Build WSGI app with middleware (same as runsimple but without print)
|
||||
func = web.httpserver.StaticMiddleware(app.wsgifunc())
|
||||
func = web.httpserver.LogMiddleware(func)
|
||||
@@ -471,6 +476,37 @@ class SchedulerHandler:
|
||||
return json.dumps({"status": "error", "message": str(e)})
|
||||
|
||||
|
||||
class HistoryHandler:
|
||||
def GET(self):
|
||||
"""
|
||||
Return paginated conversation history for a session.
|
||||
|
||||
Query params:
|
||||
session_id (required)
|
||||
page int, default 1 (1 = most recent messages)
|
||||
page_size int, default 20
|
||||
"""
|
||||
web.header('Content-Type', 'application/json; charset=utf-8')
|
||||
web.header('Access-Control-Allow-Origin', '*')
|
||||
try:
|
||||
params = web.input(session_id='', page='1', page_size='20')
|
||||
session_id = params.session_id.strip()
|
||||
if not session_id:
|
||||
return json.dumps({"status": "error", "message": "session_id required"})
|
||||
|
||||
from agent.memory import get_conversation_store
|
||||
store = get_conversation_store()
|
||||
result = store.load_history_page(
|
||||
session_id=session_id,
|
||||
page=int(params.page),
|
||||
page_size=int(params.page_size),
|
||||
)
|
||||
return json.dumps({"status": "success", **result}, ensure_ascii=False)
|
||||
except Exception as e:
|
||||
logger.error(f"[WebChannel] History API error: {e}")
|
||||
return json.dumps({"status": "error", "message": str(e)})
|
||||
|
||||
|
||||
class LogsHandler:
|
||||
def GET(self):
|
||||
"""Stream the last N lines of run.log as SSE, then tail new lines."""
|
||||
|
||||
@@ -160,7 +160,8 @@ available_setting = {
|
||||
# chatgpt指令自定义触发词
|
||||
"clear_memory_commands": ["#清除记忆"], # 重置会话指令,必须以#开头
|
||||
# channel配置
|
||||
"channel_type": "", # 通道类型,支持:{wx,wxy,terminal,wechatmp,wechatmp_service,wechatcom_app,dingtalk}
|
||||
"channel_type": "", # 通道类型,支持多渠道同时运行。单个: "feishu",多个: "feishu, dingtalk" 或 ["feishu", "dingtalk"]。可选值: web,feishu,dingtalk,wechatmp,wechatmp_service,wechatcom_app
|
||||
"web_console": True, # 是否自动启动Web控制台(默认启动)。设为False可禁用
|
||||
"subscribe_msg": "", # 订阅消息, 支持: wechatmp, wechatmp_service, wechatcom_app
|
||||
"debug": False, # 是否开启debug模式,开启后会打印更多日志
|
||||
"appdata_dir": "", # 数据目录
|
||||
@@ -186,6 +187,7 @@ available_setting = {
|
||||
"linkai_api_key": "",
|
||||
"linkai_app_code": "",
|
||||
"linkai_api_base": "https://api.link-ai.tech", # linkAI服务地址
|
||||
"cloud_host": "client.link-ai.tech",
|
||||
"minimax_api_key": "",
|
||||
"Minimax_group_id": "",
|
||||
"Minimax_base_url": "",
|
||||
|
||||
Reference in New Issue
Block a user