DeepSeek 的 Agent 公式：Model + Harness = Agent，工程外壳才是关键

"""
minimal_harness.py — 把裸模型包成 Agent 的最小 Harness
依赖: pip install openai
运行: python minimal_harness.py
"""

import json
import time
import os
from openai import OpenAI

# ---------- 1. 工具注册层 ----------
# Harness 的第一件事：声明 Agent 能用什么工具，参数长什么样

TOOLS = {
    "get_weather": {
        "description": "查询指定城市的当前天气",
        "parameters": {
            "city": {"type": "string", "description": "城市名，如 '北京'"},
        },
        "timeout": 5,  # 超时熔断阈值（秒）
    },
    "search_docs": {
        "description": "在内部文档库中搜索关键词",
        "parameters": {
            "query": {"type": "string", "description": "搜索关键词"},
            "limit": {"type": "integer", "description": "返回条数，默认5"},
        },
        "timeout": 10,
    },
}

# ---------- 2. 工具执行层 ----------
# Harness 的第二件事：真正执行工具调用，加上校验和熔断

def execute_tool(name: str, params: dict) -> dict:
    """执行工具调用，带参数校验 + 超时熔断"""
    if name not in TOOLS:
        return {"error": f"未知工具: {name}"}

    tool_spec = TOOLS[name]
    # 参数校验：只允许声明过的参数通过
    allowed_keys = set(tool_spec["parameters"].keys())
    filtered = {k: v for k, v in params.items() if k in allowed_keys}
    # 缺失必要参数则补默认值或报错
    for k, spec in tool_spec["parameters"].items():
        if k not in filtered:
            if spec["type"] == "integer":
                filtered[k] = 5  # 简单默认值
            else:
                return {"error": f"缺少必要参数: {k}"}

    # 模拟执行（真实场景替换为实际 API 调用）
    start = time.time()
    try:
        # 这里用 mock 数据演示，真实 Harness 会调外部服务
        if name == "get_weather":
            result = {"city": filtered["city"], "temp": "22°C", "condition": "晴"}
        elif name == "search_docs":
            result = {"matches": [f"文档{i}: 包含关键词'{filtered['query']}'" for i in range(filtered.get("limit", 5))]}
        else:
            result = {"error": "未实现"}

        elapsed = time.time() - start
        if elapsed > tool_spec["timeout"]:
            return {"error": f"工具 {name} 超时 ({elapsed:.1f}s > {tool_spec['timeout']}s)"}
        return result
    except Exception as e:
        return {"error": f"工具 {name} 执行异常: {e}"}

# ---------- 3. 会话记忆层 ----------
# Harness 的第三件事：管理上下文，防止 token 爆炸

MAX_HISTORY = 10  # 保留最近 10 轮对话

def manage_history(history: list, new_msg: dict) -> list:
    """追加消息并裁剪历史，防止上下文过长"""
    history.append(new_msg)
    if len(history) > MAX_HISTORY:
        # 简单策略：保留 system + 最近消息
        # 更成熟的 Harness 会做摘要压缩
        history = history[-MAX_HISTORY:]
    return history

# ---------- 4. Agent 主循环 ----------
# Harness 的核心：把模型推理和工具执行串成闭环

SYSTEM_PROMPT = """你是一个助手，可以调用工具帮助用户。
当需要外部信息时，你必须通过工具获取，不要凭猜测回答。
可用工具: {tool_names}
调用工具时严格按参数规范传值。"""

def build_tool_schema():
    """把 TOOLS 注册表转成 OpenAI function calling 格式"""
    functions = []
    for name, spec in TOOLS.items():
        func = {
            "name": name,
            "description": spec["description"],
            "parameters": {
                "type": "object",
                "properties": spec["parameters"],
                "required": [k for k in spec["parameters"]],
            },
        }
        functions.append(func)
    return functions

def run_agent(user_input: str, history: list) -> str:
    """Agent 主循环：推理 → 工具调用 → 再推理，最多 3 轮"""
    client = OpenAI(
        api_key=os.environ.get("DEEPSEEK_API_KEY", "sk-xxx"),
        base_url=os.environ.get("DEEPSEEK_BASE_URL", "https://api.deepseek.com"),
    )

    tool_schema = build_tool_schema()
    system_msg = SYSTEM_PROMPT.format(
        tool_names=", ".join(TOOLS.keys())
    )

    history = manage_history(history, {"role": "user", "content": user_input})

    max_rounds = 3  # 防止无限循环
    for _ in range(max_rounds):
        messages = [{"role": "system", "content": system_msg}] + history
        resp = client.chat.completions.create(
            model="deepseek-chat",
            messages=messages,
            functions=tool_schema,
            function_call="auto",
        )

        msg = resp.choices[0].message

        # 模型选择直接回复 → 结束
        if msg.function_call is None:
            history = manage_history(history, {"role": "assistant", "content": msg.content})
            return msg.content

        # 模型选择调用工具 → 执行 → 把结果喂回模型
        func_name = msg.function_call.name
        try:
            func_args = json.loads(msg.function_call.arguments)
        except json.JSONDecodeError:
            func_args = {}

        tool_result = execute_tool(func_name, func_args)

        # 记录工具调用和结果
        history = manage_history(history, {
            "role": "assistant",
            "content": None,
            "function_call": {
                "name": func_name,
                "arguments": msg.function_call.arguments,
            },
        })
        history = manage_history(history, {
            "role": "function",
            "name": func_name,
            "content": json.dumps(tool_result, ensure_ascii=False),
        })

    return "Agent 达到最大推理轮数，终止。"

# ---------- 5. 运行 ----------
if __name__ == "__main__":
    history = []
    print("=== 最小 Harness Agent Demo ===")
    print("输入问题，Ctrl+C 退出\n")

    while True:
        try:
            user_input = input("你: ")
            if not user_input.strip():
                continue
            answer = run_agent(user_input, history)
            print(f"Agent: {answer}\n")
        except KeyboardInterrupt:
            print("\n退出。")
            break

export DEEPSEEK_API_KEY="你的 API Key"
# 如果用 DeepSeek 官方 API，base_url 不用改
# 如果用其他兼容端点，按需设置
export DEEPSEEK_BASE_URL="https://api.deepseek.com"

pip install openai
python minimal_harness.py