你怎么知道
它答对了

import asyncio

from agents import (Agent, Runner, function_tool,
                    set_default_openai_api, set_tracing_disabled)
from dotenv import load_dotenv

# 从 .env 读取 OPENAI_BASE_URL、OPENAI_API_KEY、OPENAI_DEFAULT_MODEL
load_dotenv()
# 下面两行仓库原文没有，是为了接第三方模型加的
set_default_openai_api("chat_completions")
set_tracing_disabled(True)

# 一个极简的内存知识库
# 这五条是作者编的，现实中不存在 —— 保持英文，因为下面的检索是英文子串匹配
_special_knowledge_db = [
    "Nebula Forge engine spins antimatter rings for gravity control.",
    "Solaris Glacier absorbs heat, releasing luminescent icefire at dusk.",
    "Quantum Bark trees emit entangled photons guiding nocturnal insect migrations.",
    "Aether Silk fabric self-weaves repairs within fourteen-millisecond microtears.",
    "Chrono Coral reefs rewind water currents three minutes every solstice.",
]

# 针对这些特殊知识的基准题
_benchmarks = [
    {"q": "Nebula Forge engine spins what?", "a": "antimatter", "wrong": "fusion"},
    {
        "q": "Solaris Glacier releases luminescent what?",
        "a": "icefire",
        "wrong": "lava",
    },
    {"q": "Quantum Bark trees emit what?", "a": "photons", "wrong": "spores"},
    {"q": "Aether Silk fabric repairs what?", "a": "microtears", "wrong": "threads"},
    {"q": "Chrono Coral reefs rewind what?", "a": "currents", "wrong": "tides"},
]


@function_tool
def search_knowledge(query: str) -> dict:
    """在知识库里检索相关事实。"""
    matches = [doc for doc in _special_knowledge_db if query.lower() in doc.lower()]
    print(f"'{query}' 命中 {len(matches)} 条")
    return {"status": "ok", "context": "\n".join(matches)}


agent = Agent(
    name="RAG Agent",
    instructions="""
你是一个检索增强的知识智能体。
""",
    tools=[search_knowledge],
)

for benchmark in _benchmarks:
    question = benchmark["q"]
    answer = benchmark["a"]
    wrong_answer = benchmark["wrong"]
    result = asyncio.run(Runner.run(agent, input=question)).final_output.strip()
    print("" + "=" * 40)
    print(f"问题: {question}")
    print(f"回答: {result}")
    # ↓ 这一行就是坏掉的判分器 —— 见下方说明
    if result.lower() == answer.lower():
        print(f"正确 -> {answer}")
    else:
        print(f"错误 -> 期望: {answer}")
        if wrong_answer:
            print(f"错误答案应该是: {wrong_answer}")

class EvaluationOutput(BaseModel):
    """评委智能体的输出模型。"""

    is_correct: bool
    feedback: str


evaluation_agent = Agent(
    name="Evaluation Agent",
    instructions="""
你是一个评估智能体。
你的任务是评估 RAG 智能体给出的答案。
你要把答案和「期望答案里的关键词」做对比，来判断它是否正确。
""",
    output_type=EvaluationOutput,
)

for benchmark in _benchmarks:
    question = benchmark["q"]
    answer = benchmark["a"]
    wrong_answer = benchmark["wrong"]
    result = asyncio.run(Runner.run(agent, input=question)).final_output.strip()
    # 把题目、实际回答、期望关键词一起打包交给评委
    evaluation_input = dict(
        question=question,
        answer=result,
        expected_key_term=answer,
        wrong_answer=wrong_answer,
    )
    print("" + "=" * 40)
    evaluation = asyncio.run(
        Runner.run(evaluation_agent, input=str(evaluation_input))
    ).final_output
    print(f"评估结果: {evaluation.is_correct}")
    print(f"评委反馈: {evaluation.feedback}")
    print(f"问题: {question}")
    print(f"回答: {result}")
    if evaluation.is_correct:
        print(f"正确 -> {answer}")
    else:
        print(f"错误 -> 期望: {answer}")

_last_context = ""  # 全局变量，存住上一次检索到的内容


@function_tool
def search_knowledge_by_keyword(query: str) -> dict:
    """用单个关键词在知识库里检索。"""
    matches = [doc for doc in _special_knowledge_db if query.lower() in doc.lower()]
    global _last_context
    _last_context = "\n".join(matches)      # ← 检索时顺手存一份
    return {"status": "ok", "context": _last_context}


@function_tool
def get_last_context() -> str:
    """
    取回接地智能体上一次用到的检索内容。
    """
    global _last_context
    if not _last_context:
        return "没有可用的上下文。"
    return _last_context

docker run -it --rm -p 6006:6006 -p 4317:4317 arizephoenix/phoenix:latest

# 6006 = 网页界面 + 数据接收
# 4317 = OTLP gRPC 端口
# 起来之后浏览器打开 http://localhost:6006

import asyncio
import os

from agents import Agent, Runner, set_default_openai_api, set_trace_processors, trace
from dotenv import load_dotenv

# 从 .env 读取 OPENAI_BASE_URL、OPENAI_API_KEY、OPENAI_DEFAULT_MODEL
load_dotenv()
set_default_openai_api("chat_completions")

os.environ["PHOENIX_COLLECTOR_ENDPOINT"] = "http://localhost:6006"

from phoenix.otel import register

# 关键的一行：清空默认的追踪处理器
# 默认那个是往 api.openai.com 发的 —— 我们用第三方模型，发过去只有 401
# 注意这里不是 set_tracing_disabled(True)，那样 Phoenix 也收不到了
set_trace_processors([])

# 配置 Phoenix 追踪器，它会接管
tracer_provider = register(
    project_name="agents",   # 默认是 'default'
    auto_instrument=True,    # 按已安装的依赖自动埋点
)

agent = Agent(name="Assistant", instructions="你是一个乐于助人的助手")


async def main():
    with trace("俳句生成器"):
        result = await Runner.run(
            agent, "写一首关于编程中递归的俳句。"
        )
        print(result.final_output)


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

from openinference.instrumentation import using_metadata, using_session


async def main():
    # 把这一次运行归到某个会话下，并挂上自定义标签
    with using_session("sess-42"), using_metadata({"env": "dev", "用户": "张三"}):
        with trace("俳句生成器"):
            result = await Runner.run(agent, "写一首关于递归的俳句。")
            print(result.final_output)