126 lines
4.3 KiB
Python
126 lines
4.3 KiB
Python
from __future__ import annotations
|
||
|
||
import argparse
|
||
from pathlib import Path
|
||
|
||
from agent import CoreAgent
|
||
from core_agent.config import apply_compat_env_aliases, build_core_agent_config, load_core_agent_env, require_model_config
|
||
from providers.openai_compatible import OpenAICompatibleProvider
|
||
from providers.rule_based import RuleBasedMeetingProvider
|
||
from tools.default_tools import build_default_registry
|
||
|
||
|
||
class TerminalRenderer:
|
||
def __init__(self, assistant_name: str) -> None:
|
||
self.assistant_name = assistant_name
|
||
|
||
def render_event(self, event) -> None:
|
||
if event.type == "round_start":
|
||
print(f"\n[step {event.iteration}/{event.max_iterations}]", end="")
|
||
elif event.type == "tool_call":
|
||
print(f"\n[tool] {event.tool_name} {event.tool_args}")
|
||
elif event.type == "tool_result":
|
||
print(f"[tool-result] {_compact(event.tool_result)}")
|
||
elif event.type == "final":
|
||
print(f"\n{self.assistant_name}> {event.final_response}")
|
||
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
parser = argparse.ArgumentParser(description="Meeting memory agent CLI.")
|
||
parser.add_argument("--data-dir", default="data", help="会议记忆与原文归档保存目录。")
|
||
parser.add_argument("--once", default="", help="只执行一条消息，适合测试。")
|
||
parser.add_argument("--offline", action="store_true", help="只用于本地演示：强制使用离线规则 provider，不调用大模型。")
|
||
parser.add_argument("--list-models", action="store_true", help="列出当前 OpenAI-compatible 服务暴露的模型名。")
|
||
return parser
|
||
|
||
|
||
def main() -> None:
|
||
args = build_parser().parse_args()
|
||
workspace = Path.cwd()
|
||
load_core_agent_env(workspace)
|
||
apply_compat_env_aliases()
|
||
config = build_core_agent_config()
|
||
|
||
if args.list_models:
|
||
_list_models(config)
|
||
return
|
||
|
||
try:
|
||
provider = _build_provider(config, force_offline=args.offline)
|
||
except RuntimeError as exc:
|
||
print(f"配置错误：{exc}")
|
||
return
|
||
|
||
registry = build_default_registry(workspace / args.data_dir)
|
||
agent = CoreAgent(
|
||
provider=provider,
|
||
workspace=workspace,
|
||
skill_dirs=[workspace / "skills"],
|
||
tool_registry=registry,
|
||
max_iterations=config.max_iterations,
|
||
)
|
||
session = agent.new_session(session_id="cli_default")
|
||
renderer = TerminalRenderer(config.agent_name)
|
||
if args.once:
|
||
for event in session.stream_ask(args.once):
|
||
renderer.render_event(event)
|
||
return
|
||
|
||
print("Meeting Memory Interactive Agent")
|
||
print(r"示例：导入 D:\github_project\my_code\meeting_agent\examples\huiyi.txt")
|
||
print("示例：最近一次会议提到了哪些待办？")
|
||
print("输入 exit / quit 退出。")
|
||
while True:
|
||
try:
|
||
user_message = input("\n你> ").strip()
|
||
except (EOFError, KeyboardInterrupt):
|
||
print()
|
||
break
|
||
if user_message.lower() in {"exit", "quit", "q"}:
|
||
break
|
||
if not user_message:
|
||
continue
|
||
for event in session.stream_ask(user_message):
|
||
renderer.render_event(event)
|
||
|
||
|
||
def _build_provider(config, *, force_offline: bool):
|
||
if force_offline:
|
||
return RuleBasedMeetingProvider()
|
||
require_model_config(config)
|
||
return OpenAICompatibleProvider(
|
||
model=config.model,
|
||
api_key=config.api_key,
|
||
base_url=config.base_url,
|
||
timeout=config.timeout,
|
||
temperature=config.temperature,
|
||
)
|
||
|
||
|
||
def _list_models(config) -> None:
|
||
from openai import OpenAI
|
||
|
||
if not config.api_key or not config.base_url:
|
||
print("配置错误：列出模型需要 OPENAI_API_KEY 和 OPENAI_BASE_URL。")
|
||
return
|
||
try:
|
||
client = OpenAI(api_key=config.api_key, base_url=config.base_url, timeout=config.timeout)
|
||
models = client.models.list()
|
||
except Exception as exc:
|
||
print(f"获取模型列表失败：{exc}")
|
||
return
|
||
print(f"OPENAI_BASE_URL={config.base_url}")
|
||
print("可用模型：")
|
||
for model in models.data:
|
||
print(f"- {model.id}")
|
||
|
||
|
||
def _compact(text: str, limit: int = 240) -> str:
|
||
text = " ".join(text.split())
|
||
return text if len(text) <= limit else text[:limit] + "..."
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
|