从零开始构建AI Agent评估体系：12种LangSmith评估方法详解

 import os  
 from langchain_openai import ChatOpenAI  
 import langsmith  
   
 # 设置 LangSmith 端点（使用云版本时请勿修改）  
 os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"  
   
 # 配置 LangSmith API 密钥  
 os.environ["LANGCHAIN_API_KEY"] = "YOUR_LANGSMITH_API_KEY"  
   
 # 配置 OpenAI API 密钥  
 os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"

 # 初始化 LangSmith 客户端  
 client = langsmith.Client()

 # 创建数据集作为问答示例的容器  
 ds = client.create_dataset(  
     dataset_name=dataset_name,  
     description="A dataset for simple exact match questions."  
 )  
   
 # 每个示例由输入字典和相应的输出字典组成
 # 输入和输出在独立列表中提供，维持相同的顺序  
 client.create_examples(  
     # 输入列表，每个输入都是字典格式  
     inputs=[  
         {  
             "prompt_template": "State the year of the declaration of independence. Respond with just the year in digits, nothing else"  
         },  
         {  
             "prompt_template": "What's the average speed of an unladen swallow?"  
         },  
     ],  
     # 对应的输出列表  
     outputs=[  
         {"output": "1776"},  # 第一个提示的预期输出  
         {"output": "5"}      # 第二个提示的预期输出（陷阱问题）  
     ],  
     # 示例将被添加到的数据集 ID  
     dataset_id=ds.id,  
 )

 # 定义待测试模型  
 model = "gpt-3.5-turbo"  
   
 # 被测试系统：接收输入字典，调用指定的 ChatOpenAI 模型，返回字典格式的输出  
 def predict_result(input_: dict) -> dict:  
     # 输入字典包含 "prompt_template" 键，与数据集输入中定义的键一致  
     prompt = input_["prompt_template"]  
       
     # 初始化并调用模型  
     response = ChatOpenAI(model=model, temperature=0).invoke(prompt)  
       
     # 输出键 "output" 与数据集输出中的键匹配，用于比较  
     return {"output": response.content}

 from langsmith.evaluation import EvaluationResult, run_evaluator  
   
 # @run_evaluator 装饰器将函数注册为自定义评估器  
 @run_evaluator  
 def compare_label(run, example) -> EvaluationResult:  
     """
     用于检查精确匹配的自定义评估器
       
     Args:  
         run: LangSmith 运行对象，包含模型输出  
         example: LangSmith 示例对象，包含参考数据  
       
     Returns:  
         包含键和分数的 EvaluationResult 对象  
     """  
     # 从运行输出字典获取模型预测
     # 键 'output' 必须与 `predict_result` 函数返回内容匹配  
     prediction = run.outputs.get("output") or ""  
       
     # 从示例输出字典获取参考答案
     # 键 'output' 必须与数据集中定义内容匹配  
     target = example.outputs.get("output") or ""  
       
     # 执行比较操作  
     match = prediction == target  
       
     # 返回结果，键值为结果中分数的命名方式
     # 精确匹配分数通常为二进制（匹配为 1，不匹配为 0）  
     return EvaluationResult(key="matches_label", score=int(match))

 from langchain.smith import RunEvalConfig  
   
 # 定义评估运行的配置  
 eval_config = RunEvalConfig(  
     # 通过字符串名称指定内置评估器  
     evaluators=["exact_match"],   
       
     # 在列表中直接传递自定义评估器函数  
     custom_evaluators=[compare_label],  
 )  
   
 # 触发评估执行
 # 对数据集中每个示例运行 `predict_result` 函数
 # 然后使用 `eval_config` 中的评估器对结果评分  
 client.run_on_dataset(  
     dataset_name=dataset_name,  
     llm_or_chain_factory=predict_result,  
     evaluation=eval_config,  
     verbose=True, # 打印进度条和链接  
     project_metadata={"version": "1.0.1", "model": model}, # 项目可选元数据  
 )

 View the evaluation results for project 'gregarious-doctor-77' at:  
 https://smith.langchain.com/o/your-org-id/datasets/some-dataset-uuid/compare?selectedSessions=some-session-uuid  
   
 View all tests for Dataset Oracle of Exactness at:  
 https://smith.langchain.com/o/your-org-id/datasets/some-dataset-uuid  
 [------------------------------------------------->] 2/2

 # 在 LangSmith 中创建数据集  
 dataset = client.create_dataset(  
     dataset_name=dataset_name,  
     description="Q&A dataset about LangSmith documentation."  
 )  
   
 # 问答示例，答案作为'标准答案'  
 qa_examples = [  
     (  
         "What is LangChain?",  
         "LangChain is an open-source framework for building applications using large language models. It is also the name of the company building LangSmith.",  
     ),  
     (  
         "How might I query for all runs in a project?",  
         "You can use client.list_runs(project_name='my-project-name') in Python, or client.ListRuns({projectName: 'my-project-name'}) in TypeScript.",  
     ),  
     (  
         "What's a langsmith dataset?",  
         "A LangSmith dataset is a collection of examples. Each example contains inputs and optional expected outputs or references for that data point.",  
     ),  
     (  
         "How do I move my project between organizations?",  
         "LangSmith doesn't directly support moving projects between organizations.",  
     ),  
 ]  
   
 # 将示例添加到数据集
 # 输入键为 'question'，输出键为 'answer'
 # 这些键必须与 RAG 链期望和生成的内容匹配  
 for question, answer in qa_examples:  
     client.create_example(  
         inputs={"question": question},  
         outputs={"answer": answer},  
         dataset_id=dataset.id,  
     )

 from langchain_community.document_loaders import RecursiveUrlLoader  
 from langchain_community.document_transformers import Html2TextTransformer  
 from langchain_community.vectorstores import Chroma  
 from langchain_text_splitters import TokenTextSplitter  
 from langchain_openai import OpenAIEmbeddings  
   
 # 1. 从网络加载文档  
 api_loader = RecursiveUrlLoader("https://docs.smith.langchain.com")  
 raw_documents = api_loader.load()  
   
 # 2. 将 HTML 转换为干净文本并分割为可管理的片段  
 doc_transformer = Html2TextTransformer()  
 transformed = doc_transformer.transform_documents(raw_documents)  
 text_splitter = TokenTextSplitter(model_name="gpt-3.5-turbo", chunk_size=2000, chunk_overlap=200)  
 documents = text_splitter.split_documents(transformed)  
   
 # 3. 创建向量存储检索器  
 embeddings = OpenAIEmbeddings()  
 vectorstore = Chroma.from_documents(documents, embeddings)  
 retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

 from operator import itemgetter  
 from langchain_core.output_parsers import StrOutputParser  
 from langchain_core.prompts import ChatPromptTemplate  
   
 # 定义发送给 LLM 的提示模板  
 prompt = ChatPromptTemplate.from_messages(  
     [  
         (  
             "system",  
             "You are a helpful documentation Q&A assistant, trained to answer"  
             " questions from LangSmith's documentation."  
             " LangChain is a framework for building applications using large language models."  
             "\nThe current time is {time}.\n\nRelevant documents will be retrieved in the following messages.",  
         ),  
         ("system", "{context}"), # 检索文档的占位符  
         ("human", "{question}"),  # 用户问题的占位符  
     ]  
 ).partial(time=str(datetime.now()))  
   
 # 初始化 LLM，使用大上下文窗口和低温度参数以获得更准确的响应  
 model = ChatOpenAI(model="gpt-3.5-turbo-16k", temperature=0)  
   
 # 定义生成链，将提示传递给模型再传递给输出解析器  
 response_generator = prompt | model | StrOutputParser()

 # 配置评估使用"qa"评估器，基于参考答案对"正确性"进行评分  
 eval_config = RunEvalConfig(  
     evaluators=["qa"],  
 )  
   
 # 在数据集上运行 RAG 链并应用评估器  
 client.run_on_dataset(  
     dataset_name=dataset_name,  
     llm_or_chain_factory=rag_chain,  
     evaluation=eval_config,  
     verbose=True,  
     project_metadata={"version": "1.0.0", "model": "gpt-3.5-turbo"},  
 )

 View the evaluation results for project 'witty-scythe-29' at:  
 https://smith.langchain.com/o/your-org-id/datasets/some-dataset-uuid/compare?selectedSessions=some-session-uuid  
   
 View all tests for Dataset Retrieval QA - LangSmith Docs at:  
 https://smith.langchain.com/o/your-org-id/datasets/some-dataset-uuid  
 [------------------------------------------------->] 5/5

 # 定义改进的提示模板  
 prompt_v2 = ChatPromptTemplate.from_messages(  
     [  
         (  
             "system",  
             "You are a helpful documentation Q&A assistant, trained to answer"  
             " questions from LangSmith's documentation."  
             "\nThe current time is {time}.\n\nRelevant documents will be retrieved in the following messages.",  
         ),  
         ("system", "{context}"),  
         ("human", "{question}"),  
         # 新增防止幻觉的指令  
         (  
             "system",  
             "Respond as best as you can. If no documents are retrieved or if you do not see an answer in the retrieved documents,"  
             " admit you do not know or that you don't see it being supported at the moment.",  
         ),  
     ]  
 ).partial(time=lambda: str(datetime.now()))

 # LangSmith 上公共数据集的 URL  
 dataset_url = "https://smith.langchain.com/public/08ab7912-006e-4c00-a973-0f833e74907b/d"  
 dataset_name = "Contract Extraction Eval Dataset"  
   
 # 将公共数据集克隆到账户中  
 client.clone_public_dataset(dataset_url, dataset_name=dataset_name)

 from typing import List, Optional  
 from pydantic import BaseModel  
   
 # 定义当事方地址的模式  
 class Address(BaseModel):  
     street: str  
     city: str  
     state: str  
   
 # 定义合同中当事方的模式  
 class Party(BaseModel):  
     name: str  
     address: Address  
   
 # 整个合同的顶级模式  
 class Contract(BaseModel):  
     document_title: str  
     effective_date: str  
     parties: List[Party]

 from langchain.chains import create_extraction_chain  
 from langchain_anthropic import ChatAnthropic  
   
 # 使用能够遵循复杂指令的强大模型
 # 注：可以替换为等效的 OpenAI 模型  
 llm = ChatAnthropic(model="claude-2.1", temperature=0, max_tokens=4000)  
   
 # 创建提取链，提供模式和 LLM  
 extraction_chain = create_extraction_chain(Contract.schema(), llm)

 from langsmith.evaluation import LangChainStringEvaluator  
   
 # 评估配置指定 JSON 感知评估器
 # 'json_edit_distance' 评估器比较两个 JSON 对象的结构和内容  
 eval_config = RunEvalConfig(  
     evaluators=[  
         LangChainStringEvaluator("json_edit_distance")  
     ]  
 )  
   
 # 运行评估  
 client.run_on_dataset(  
     dataset_name=dataset_name,  
     llm_or_chain_factory=extraction_chain,  
     evaluation=eval_config,  
     # 数据集中的输入键为'context'，映射到链的'input'键  
     input_mapper=lambda x: {"input": x["context"]},  
     # 链的输出为字典 {'text': [...]}，关注'text'值  
     output_mapper=lambda x: x['text'],  
     verbose=True,  
     project_metadata={"version": "1.0.0", "model": "claude-2.1"},  
 )

 # 问题列表和相应的 pandas 代码以找到答案  
 questions_with_references = [  
     ("How many passengers were on the Titanic?", "len(df)"),  
     ("How many passengers survived?", "df['Survived'].sum()"),  
     ("What was the average age of the passengers?", "df['Age'].mean()"),  
     ("How many male and female passengers were there?", "df['Sex'].value_counts()"),  
     ("What was the average fare paid for the tickets?", "df['Fare'].mean()"),  
 ]  
   
 # 创建唯一数据集名称  
 dataset_name = "Dynamic Titanic QA"  
   
 # 在 LangSmith 中创建数据集  
 dataset = client.create_dataset(  
     dataset_name=dataset_name,  
     description="QA over the Titanic dataset with dynamic references.",  
 )  
   
 # 填充数据集，输入为问题，输出为代码  
 client.create_examples(  
     inputs=[{"question": q} for q, r in questions_with_references],  
     outputs=[{"reference_code": r} for q, r in questions_with_references],  
     dataset_id=dataset.id,  
 )

 import pandas as pd  
   
 # 从 URL 加载泰坦尼克号数据集  
 titanic_url = "https://raw.githubusercontent.com/jorisvandenbossche/pandas-tutorial/master/data/titanic.csv"  
 df = pd.read_csv(titanic_url)

 # 为 Agent 定义 LLM  
 llm = ChatOpenAI(model="gpt-4", temperature=0)  
   
 # 此函数在 `df` 当前状态下创建和调用 Agent  
 def predict_pandas_agent(inputs: dict):  
     # Agent 使用当前 `df` 创建  
     agent = create_pandas_dataframe_agent(agent_type="openai-tools", llm=llm, df=df)  
     return agent.invoke({"input": inputs["question"]})

 from typing import Optional  
 from langchain.evaluation.criteria.eval_chain import LabeledCriteriaEvalChain  
   
 class DynamicReferenceEvaluator(LabeledCriteriaEvalChain):  
     def _get_eval_input(  
         self,  
         prediction: str,  
         reference: Optional[str],  
         input: Optional[str],  
     ) -> dict:  
         # 从父类获取标准输入字典  
         eval_input = super()._get_eval_input(prediction, reference, input)  
 
         # 这里的'reference'是代码片段，例如"len(df)"
         # 执行它以获得实时标准答案值
         # 警告：使用 `eval` 可能有风险，仅运行可信代码  
         live_ground_truth = eval(eval_input["reference"])  
           
         # 用实际的实时答案替换代码片段  
         eval_input["reference"] = str(live_ground_truth)  
           
         return eval_input

 # 创建自定义评估器链的实例  
 base_evaluator=DynamicReferenceEvaluator.from_llm(  
     criteria="correctness", llm=ChatOpenAI(model="gpt-4", temperature=0)  
 )  
   
 # 将其包装在 LangChainStringEvaluator 中以正确映射运行/示例字段  
 dynamic_evaluator=LangChainStringEvaluator(  
     base_evaluator,  
     # 此函数将数据集字段映射到评估器期望的内容  
     prepare_data=lambdarun, example: {  
         "prediction": run.outputs["output"],  
         "reference": example.outputs["reference_code"],  
         "input": example.inputs["question"],  
     },  
 )  
   
 # 在时间"T1"运行评估  
 client.run_on_dataset(  
     dataset_name=dataset_name,  
     llm_or_chain_factory=predict_pandas_agent,  
     evaluation=RunEvalConfig(  
         custom_evaluators=[dynamic_evaluator],  
     ),  
     project_metadata={"time": "T1"},  
     max_concurrency=1, # Pandas Agent 不是线程安全的  
 )

 # 通过数据加倍来模拟数据更新  
 df_doubled = pd.concat([df, df], ignore_index=True)  
 df = df_doubled

 # 在时间"T2"对更新数据重新运行评估  
 client.run_on_dataset(  
     dataset_name=dataset_name,  
     llm_or_chain_factory=predict_pandas_agent,  
     evaluation=RunEvalConfig(  
         custom_evaluators=[dynamic_evaluator],  
     ),  
     project_metadata={"time": "T2"},  
     max_concurrency=1,  
 )

 # 问题列表，每个都有参考答案和预期工具轨迹  
 agent_questions = [  
     (  
         "Why was a $10 calculator app a top-rated Nintendo Switch game?",  
         {  
             "reference": "It became an internet meme due to its high price point.",  
             "expected_steps": ["duck_duck_go"], # 期望网络搜索  
         },  
     ),  
     (  
         "hi",  
         {  
             "reference": "Hello, how can I assist you?",  
             "expected_steps": [],  # 期望无工具调用的直接响应  
         },  
     ),  
     (  
         "What's my first meeting on Friday?",  
         {  
             "reference": 'Your first meeting is 8:30 AM for "Team Standup"',  
             "expected_steps": ["check_calendar"],  # 期望日历工具  
         },  
     ),  
 ]  
   
 # 在 LangSmith 中创建数据集  
 dataset_name = "Agent Trajectory Eval"  
 dataset = client.create_dataset(  
     dataset_name=dataset_name,  
     description="Dataset for evaluating agent tool use and trajectory.",  
 )  
   
 # 使用输入和多部分输出填充数据集  
 client.create_examples(  
     inputs=[{"question": q[0]} for q in agent_questions],  
     outputs=[q[1] for q in agent_questions],  
     dataset_id=dataset.id,  
 )

 from langchain.agents import AgentExecutor, create_openai_tools_agent  
 from langchain_community.tools import DuckDuckGoSearchResults  
 from langchain_core.prompts import MessagesPlaceholder  
 from langchain_core.tools import tool  
   
 # 演示用的模拟工具  
 @tool  
 def check_calendar(date: str) -> list:  
     """检查用户在指定日期的日历会议"""  
     if "friday" in date.lower():  
         return 'Your first meeting is 8:30 AM for "Team Standup"'  
     return "You have no meetings."  
   
 # 此工厂函数创建 Agent 执行器  
 def create_agent_executor(inputs: dict):  
     llm = ChatOpenAI(model="gpt-4", temperature=0)  
     tools = [DuckDuckGoSearchResults(name="duck_duck_go"), check_calendar]  
       
     prompt = ChatPromptTemplate.from_messages([  
         ("system", "You are a helpful assistant."),  
         ("user", "{question}"),  
         MessagesPlaceholder(variable_name="agent_scratchpad"),  
     ])  
       
     agent_runnable = create_openai_tools_agent(llm, tools, prompt)  
       
     # 关键步骤：`return_intermediate_steps=True` 使轨迹在输出中可用  
     executor = AgentExecutor(  
         agent=agent_runnable,  
         tools=tools,  
         return_intermediate_steps=True,  
     )  
     return executor.invoke(inputs)

 # 自定义评估器函数  
 @run_evaluator  
 def trajectory_evaluator(run: Run, example: Optional[Example] = None) -> dict:  
     # 1. 从运行输出获取 Agent 的实际工具调用
     # 'intermediate_steps' 是 (action, observation) 元组列表  
     intermediate_steps = run.outputs.get("intermediate_steps", [])  
     actual_trajectory = [action.tool for action, observation in intermediate_steps]  
       
     # 2. 从数据集示例获取预期工具调用  
     expected_trajectory = example.outputs.get("expected_steps", [])  
       
     # 3. 比较并分配二进制分数  
     score = int(actual_trajectory == expected_trajectory)  
       
     # 4. 返回结果  
     return {"key": "trajectory_correctness", "score": score}

 # 'qa' 评估器需要知道用于输入、预测和参考的字段  
 qa_evaluator = LangChainStringEvaluator(  
     "qa",  
     prepare_data=lambda run, example: {  
         "input": example.inputs["question"],  
         "prediction": run.outputs["output"],  
         "reference": example.outputs["reference"],  
     },  
 )  
   
 # 使用两个评估器运行评估  
 client.run_on_dataset(  
     dataset_name=dataset_name,  
     llm_or_chain_factory=create_agent_executor,  
     evaluation=RunEvalConfig(  
         # 包括自定义轨迹评估器和内置 QA 评估器  
         evaluators=[qa_evaluator],  
         custom_evaluators=[trajectory_evaluator],  
     ),  
     max_concurrency=1,  
 )

 # 工具选择数据集的公共 URL  
 dev_dataset_url = "https://smith.langchain.com/public/bdf7611c-3420-4c71-a492-42715a32d61e/d"  
 dataset_name = "Tool Selection (Logistics) Dev"  
   
 # 将数据集克隆到 LangSmith 账户  
 client.clone_public_dataset(dev_dataset_url, dataset_name=dataset_name)

 from langsmith.evaluation import run_evaluator  
   
 @run_evaluator  
 def selected_tools_precision(run: Run, example: Example) -> dict:  
     # 数据集中的'expected'字段包含正确工具名称  
     expected_tools = set(example.outputs["expected"][0])  
       
     # Agent 输出是预测工具调用列表  
     predicted_calls = run.outputs.get("output", [])  
     predicted_tools = {tool["type"] for tool in predicted_calls}  
       
     # 计算精度：（正确预测的工具）/（所有预测的工具）  
     if not predicted_tools:  
         score = 1 if not expected_tools else 0  
     else:  
         true_positives = predicted_tools.intersection(expected_tools)  
         score = len(true_positives) / len(predicted_tools)  
           
     return {"key": "tool_selection_precision", "score": score}

 import json  
 from langchain_core.output_parsers.openai_tools import JsonOutputToolsParser  
 from langchain_core.prompts import ChatPromptTemplate  
 from langchain_openai import ChatOpenAI  
   
 # 从本地文件加载工具规范  
 with open("./data/tools.json") as f:  
     tools = json.load(f)  
   
 # 定义提示并将工具绑定到 LLM  
 assistant_prompt = ChatPromptTemplate.from_messages([  
     ("system", "You are a helpful assistant. Respond to the user's query using the provided tools."),  
     ("user", "{query}"),  
 ])  
 llm = ChatOpenAI(model="gpt-3.5-turbo").bind_tools(tools)  
 chain = assistant_prompt | llm | JsonOutputToolsParser()

 # 使用自定义精度评估器配置评估  
 eval_config = RunEvalConfig(custom_evaluators=[selected_tools_precision])  
   
 # 运行评估  
 test_results = client.run_on_dataset(  
     dataset_name=dataset_name,  
     llm_or_chain_factory=chain,  
     evaluation=eval_config,  
     verbose=True,  
     project_metadata={"model": "gpt-3.5-turbo", "tool_variant": "original"},  
 )

 # 改进提示以纠正 Agent 工具调用  
 improver_prompt = ChatPromptTemplate.from_messages([  
     # 此处为改进提示的具体实现
 ])

 # 使用更新工具描述创建新链  
 llm_v2 = ChatOpenAI(model="gpt-3.5-turbo").bind_tools(new_tools)  
 updated_chain = assistant_prompt | llm_v2 | JsonOutputToolsParser()  
   
 # 重新运行评估  
 updated_test_results = client.run_on_dataset(  
     dataset_name=dataset_name,  
     llm_or_chain_factory=updated_chain,  
     evaluation=eval_config,  
     verbose=True,  
     project_metadata={"model": "gpt-3.5-turbo", "tool_variant": "improved"},  
 )

 # 示例数据集，每个输入都包含问题和上下文  
 examples = [
     # 具体示例数据将在此处定义
 ]  
   
 dataset_name = "RAG Faithfulness Eval"  
 dataset = client.create_dataset(dataset_name=dataset_name)  
   
 # 在 LangSmith 中创建示例，传递复杂输入/输出对象  
 client.create_examples(  
     inputs=[e["inputs"] for e in examples],  
     dataset_id=dataset.id,  
 )

 # 将单独评估的组件
 # response_synthesizer 的具体实现将在此处定义

 # 安装 RAGAS（如果尚未安装）
 # pip install ragas
 
 from ragas.langchain.evaluation import RagasEvaluatorChain
 from ragas.metrics import (
     answer_relevancy,
     faithfulness,
     context_relevancy,
     context_precision,
 )

 # 创建 RAGAS 评估数据集
 dataset_name = "RAG Evaluation with RAGAS"
 
 # 创建数据集示例，每个都包含问题、上下文和参考答案
 ragas_examples = [
     {
         "inputs": {
             "question": "What is LangSmith?",
             "contexts": [
                 "LangSmith is a platform for building production-grade LLM applications.",
                 "It provides debugging, testing, and monitoring capabilities."
             ]
         },
         "outputs": {
             "answer": "LangSmith is a platform for building production-grade LLM applications with debugging and monitoring features.",
             "ground_truth": "LangSmith is a comprehensive platform for developing and deploying LLM applications."
         }
     }
 ]
 
 # 在 LangSmith 中创建数据集
 dataset = client.create_dataset(dataset_name=dataset_name)
 client.create_examples(
     inputs=[e["inputs"] for e in ragas_examples],
     outputs=[e["outputs"] for e in ragas_examples],
     dataset_id=dataset.id,
 )

 # 创建 RAGAS 评估器
 evaluator_chains = []
 
 for metric in [answer_relevancy, faithfulness, context_relevancy, context_precision]:
     evaluator_chain = RagasEvaluatorChain(metric=metric)
     evaluator_chains.append(evaluator_chain)

 # 运行 RAGAS 评估
 eval_config = RunEvalConfig(
     custom_evaluators=evaluator_chains
 )
 
 client.run_on_dataset(
     dataset_name=dataset_name,
     llm_or_chain_factory=rag_chain,
     evaluation=eval_config,
     verbose=True,
     project_metadata={"evaluation_type": "ragas", "version": "1.0.0"},
 )

 import time
 from langsmith import traceable
 
 @traceable
 def rag_with_feedback(question: str) -> dict:
     """
     启用反馈的 RAG 函数，跟踪用户交互
     """
     # 运行 RAG 管道
     docs = retriever.invoke(question)
     answer = response_generator.invoke({
         "question": question, 
         "context": docs
     })
     
     return {
         "answer": answer,
         "source_documents": docs,
         "timestamp": time.time()
     }

 def collect_user_feedback(run_id: str, feedback_type: str, score: float = None, comment: str = None):
     """
     为特定运行收集用户反馈
     
     Args:
         run_id: LangSmith 运行 ID
         feedback_type: 反馈类型（'thumbs_up_down', 'rating', 'comment'）
         score: 数值分数（用于评分）
         comment: 文本评论（可选）
     """
     client.create_feedback(
         run_id=run_id,
         key=feedback_type,
         score=score,
         comment=comment
     )

 # 模拟用户会话
 questions = [
     "What is the capital of France?",
     "How do I reset my password?", 
     "What are the benefits of LangSmith?"
 ]
 
 run_ids = []
 
 for question in questions:
     # 运行 RAG 系统
     result = rag_with_feedback(question)
     
     # 在真实场景中，从 UI 获取 run_id
     # 为演示目的，模拟获取最新运行 ID
     run_id = "simulated_run_id"  # 实际中从 traceable 装饰器获取
     run_ids.append(run_id)
     
     # 模拟用户反馈
     # 在真实应用中，这些来自 UI 交互
     collect_user_feedback(
         run_id=run_id,
         feedback_type="rating",
         score=4.5,  # 5分制
         comment="答案很有帮助，但可以更详细"
     )

 def analyze_feedback_trends():
     """
     分析收集的反馈以识别趋势和改进领域
     """
     # 获取最近的运行和反馈
     runs = client.list_runs(
         project_name="your-project-name",
         limit=100
     )
     
     ratings = []
     comments = []
     
     for run in runs:
         # 获取此运行的反馈
         feedback = client.list_feedback(run_ids=[run.id])
         for f in feedback:
             if f.key == "rating" and f.score:
                 ratings.append(f.score)
             if f.comment:
                 comments.append(f.comment)
     
     # 计算平均评分
     avg_rating = sum(ratings) / len(ratings) if ratings else 0
     
     print(f"平均用户评分：{avg_rating:.2f}")
     print(f"总反馈数：{len(ratings)}")
     
     # 分析评论中的常见主题
     negative_feedback = [c for c in comments if "不好" in c.lower() or "错误" in c.lower()]
     
     return {
         "average_rating": avg_rating,
         "total_feedback": len(ratings),
         "negative_feedback_count": len(negative_feedback)
     }

 def ab_test_with_feedback():
     """
     使用实时反馈运行 A/B 测试来比较 Agent 版本
     """
     # 版本 A：当前 Agent
     # 版本 B：改进的 Agent（基于反馈）
     
     import random
     
     def improved_rag_with_feedback(question: str) -> dict:
         """改进版本，基于用户反馈调整了提示"""
         # 这里实施基于负面反馈的改进
         return rag_with_feedback(question)
     
     # 随机选择版本
     version = "A" if random.random() < 0.5 else "B"
     
     if version == "A":
         result = rag_with_feedback("sample question")
     else:
         result = improved_rag_with_feedback("sample question")
         
     # 使用版本标记跟踪
     # 这允许比较不同版本的性能
     return result, version

 from langsmith.evaluation import evaluate_comparative
 
 # 创建用于比较的数据集
 dataset_name = "Pairwise Comparison Dataset"
 
 comparison_questions = [
     "Explain quantum computing in simple terms",
     "What are the pros and cons of remote work?",
     "How can I improve my Python coding skills?",
     "What is the future of artificial intelligence?"
 ]
 
 # 创建数据集
 dataset = client.create_dataset(dataset_name=dataset_name)
 client.create_examples(
     inputs=[{"question": q} for q in comparison_questions],
     dataset_id=dataset.id,
 )

 # Agent A：更保守，简洁的响应
 def agent_a(inputs: dict) -> dict:
     prompt = f"请简洁地回答：{inputs['question']}"
     response = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.3).invoke(prompt)
     return {"output": response.content}
 
 # Agent B：更详细，创造性的响应
 def agent_b(inputs: dict) -> dict:
     prompt = f"请提供详细、有见地的回答：{inputs['question']}"
     response = ChatOpenAI(model="gpt-4", temperature=0.7).invoke(prompt)
     return {"output": response.content}

 @run_evaluator
 def pairwise_evaluator(run, example) -> dict:
     """
     比较两个响应并选择较好的一个
     此处模拟比较逻辑
     """
     # 在真实场景中，将有来自两个不同 Agent 的响应
     # 此处模拟比较过程
     
     judge_prompt = f"""
     比较这两个对问题 "{example.inputs['question']}" 的响应：
     
     响应 A：{run.outputs.get('output_a', '')}
     响应 B：{run.outputs.get('output_b', '')}
     
     哪个响应更好？考虑准确性、实用性和清晰度。
     响应只有 'A' 或 'B'。
     """
     
     judge = ChatOpenAI(model="gpt-4", temperature=0)
     judgment = judge.invoke(judge_prompt).content.strip()
     
     # 将判断转换为分数：A 获胜 = 1，B 获胜 = 0
     score = 1 if judgment.upper() == 'A' else 0
     
     return {"key": "agent_a_wins", "score": score}

 # 为进行成对比较，需要为每个示例生成两个响应
 def pairwise_system(inputs: dict) -> dict:
     """
     运行两个 Agent 并返回两个响应以供比较
     """
     response_a = agent_a(inputs)
     response_b = agent_b(inputs)
     
     return {
         "output_a": response_a["output"],
         "output_b": response_b["output"]
     }
 
 # 运行比较评估
 eval_config = RunEvalConfig(
     custom_evaluators=[pairwise_evaluator]
 )
 
 client.run_on_dataset(
     dataset_name=dataset_name,
     llm_or_chain_factory=pairwise_system,
     evaluation=eval_config,
     verbose=True,
     project_metadata={"evaluation_type": "pairwise_comparison"},
 )

 def analyze_pairwise_results():
     """
     分析成对比较结果以确定获胜的 Agent
     """
     # 获取评估结果
     runs = client.list_runs(project_name="your-project-name", limit=50)
     
     agent_a_wins = 0
     total_comparisons = 0
     
     for run in runs:
         feedback = client.list_feedback(run_ids=[run.id])
         for f in feedback:
             if f.key == "agent_a_wins":
                 total_comparisons += 1
                 if f.score == 1:
                     agent_a_wins += 1
     
     if total_comparisons > 0:
         win_rate_a = agent_a_wins / total_comparisons
         win_rate_b = 1 - win_rate_a
         
         print(f"Agent A 胜率：{win_rate_a:.2%}")
         print(f"Agent B 胜率：{win_rate_b:.2%}")
         
         winner = "Agent A" if win_rate_a > 0.5 else "Agent B"
         print(f"获胜者：{winner}")

 def tournament_evaluation(agents: dict):
     """
     在多个 Agent 之间运行锦标赛式评估
     
     Args:
         agents: 字典，包含 Agent 名称和函数
     """
     results = {}
     
     # 比较每对 Agent
     for name_a, agent_a in agents.items():
         for name_b, agent_b in agents.items():
             if name_a != name_b:
                 # 运行成对比较
                 comparison_key = f"{name_a}_vs_{name_b}"
                 # 实施比较逻辑...
                 results[comparison_key] = {"winner": name_a, "margin": 0.65}
     
     return results

 from typing import List, Dict
 import json
 
 class SimulatedCustomer:
     """模拟具有特定问题和个性的客户"""
     
     def __init__(self, scenario: dict):
         self.scenario = scenario
         self.issue = scenario["issue"]
         self.personality = scenario["personality"] 
         self.conversation_history = []
         self.issue_resolved = False
         
     def generate_response(self, agent_message: str) -> str:
         """基于 Agent 消息生成客户响应"""
         
         # 使用 LLM 生成现实的客户响应
         prompt = f"""
         您是一个{self.personality}的客户，有以下问题：{self.issue}
         
         对话历史：
         {self._format_history()}
         
         支持代理刚刚说："{agent_message}"
         
         作为客户，您会如何回应？保持您的个性和问题一致。
         如果您的问题得到解决，请明确表示满意。
         """
         
         llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
         response = llm.invoke(prompt).content
         
         # 检查问题是否已解决
         if "谢谢" in response.lower() and "解决" in response.lower():
             self.issue_resolved = True
             
         return response
         
     def _format_history(self) -> str:
         """格式化对话历史"""
         return "\n".join([
             f"{'客户' if msg['role'] == 'customer' else '代理'}：{msg['content']}" 
             for msg in self.conversation_history
         ])

 class CustomerSupportAgent:
     """客户支持 AI Agent"""
     
     def __init__(self):
         self.conversation_history = []
         
     def respond(self, customer_message: str, context: dict = None) -> str:
         """生成对客户消息的响应"""
         
         prompt = f"""
         您是一个有用且专业的客户支持代理。
         
         对话历史：
         {self._format_history()}
         
         客户刚刚说："{customer_message}"
         
         提供有用、准确和专业的响应。尝试解决客户的问题。
         """
         
         llm = ChatOpenAI(model="gpt-4", temperature=0.3)
         response = llm.invoke(prompt).content
         
         return response
         
     def _format_history(self) -> str:
         """格式化对话历史"""
         return "\n".join([
             f"{'客户' if msg['role'] == 'customer' else '代理'}：{msg['content']}" 
             for msg in self.conversation_history
         ])

 def run_conversation_simulation(scenario: dict, max_turns: int = 10) -> dict:
     """
     运行客户支持对话的仿真
     
     Args:
         scenario: 包含客户问题和个性的字典
         max_turns: 最大对话轮数
         
     Returns:
         包含对话历史和指标的字典
     """
     
     customer = SimulatedCustomer(scenario)
     agent = CustomerSupportAgent()
     
     conversation = []
     
     # 客户开始对话
     initial_message = f"你好，我有一个问题：{scenario['issue']}"
     conversation.append({"role": "customer", "content": initial_message})
     
     for turn in range(max_turns):
         # Agent 响应
         agent_response = agent.respond(
             conversation[-1]["content"] if conversation[-1]["role"] == "customer" else ""
         )
         conversation.append({"role": "agent", "content": agent_response})
         
         # 检查问题是否已解决
         if customer.issue_resolved:
             break
             
         # 客户响应
         customer_response = customer.generate_response(agent_response)
         conversation.append({"role": "customer", "content": customer_response})
         
         # 更新历史
         customer.conversation_history = conversation
         agent.conversation_history = conversation
     
     return {
         "conversation": conversation,
         "issue_resolved": customer.issue_resolved,
         "turns_to_resolution": len(conversation) // 2,
         "scenario": scenario
     }

 simulation_scenarios = [
     {
         "issue": "我的订单迟到了，我需要退款",
         "personality": "不耐烦和愤怒"
     },
     {
         "issue": "我忘记了我的密码，无法登录",
         "personality": "困惑但礼貌"
     },
     {
         "issue": "您的产品坏了，我想换货",
         "personality": "挫败但合理"
     },
     {
         "issue": "我想取消我的订阅",
         "personality": "坚决但友好"
     }
 ]

 def evaluate_conversation_simulations():
     """运行多个仿真并分析结果"""
     
     results = []
     
     for scenario in simulation_scenarios:
         # 运行仿真
         result = run_conversation_simulation(scenario)
         results.append(result)
         
         # 将结果记录到 LangSmith
         client.create_run(
             name="conversation_simulation",
             run_type="chain",
             inputs={"scenario": scenario},
             outputs={
                 "conversation": result["conversation"],
                 "metrics": {
                     "issue_resolved": result["issue_resolved"],
                     "turns_to_resolution": result["turns_to_resolution"]
                 }
             }
         )
     
     # 分析聚合指标
     total_simulations = len(results)
     resolved_count = sum(1 for r in results if r["issue_resolved"])
     avg_turns = sum(r["turns_to_resolution"] for r in results) / total_simulations
     
     print(f"仿真运行：{total_simulations}")
     print(f"问题解决率：{resolved_count/total_simulations:.2%}")
     print(f"平均解决轮数：{avg_turns:.1f}")
     
     return results

 @run_evaluator
 def conversation_quality_evaluator(run, example) -> dict:
     """评估整个对话的质量"""
     
     conversation = run.outputs.get("conversation", [])
     
     # 使用 LLM 评估对话质量
     prompt = f"""
     评估这个客户支持对话的质量：
     
     {json.dumps(conversation, indent=2, ensure_ascii=False)}
     
     请评分（1-10）：
     1. 专业性：代理是否保持专业语调？
     2. 有效性：代理是否有效解决了问题？
     3. 效率：对话是否简洁而重点明确？
     
     返回 JSON 格式：{{"professionalism": X, "effectiveness": X, "efficiency": X}}
     """
     
     judge = ChatOpenAI(model="gpt-4", temperature=0)
     evaluation = judge.invoke(prompt).content
     
     try:
         scores = json.loads(evaluation)
         overall_score = sum(scores.values()) / len(scores)
         return {"key": "conversation_quality", "score": overall_score}
     except:
         return {"key": "conversation_quality", "score": 5.0}  # 默认分数

 import re
 import json
 import time
 from typing import Dict, Any
 from datetime import datetime
 
 @run_evaluator
 def response_time_evaluator(run, example) -> dict:
     """测量响应生成时间"""
     
     # 计算运行持续时间
     if hasattr(run, 'start_time') and hasattr(run, 'end_time'):
         duration = (run.end_time - run.start_time).total_seconds()
     else:
         duration = 0  # 备用
         
     # 分数基于响应时间（更快 = 更好）
     # 5秒以下 = 1.0，10秒以上 = 0.0
     score = max(0, min(1, (10 - duration) / 5))
     
     return {"key": "response_time", "score": score}
 
 @run_evaluator  
 def json_validity_evaluator(run, example) -> dict:
     """检查输出是否为有效的 JSON"""
     
     output = run.outputs.get("output", "")
     
     try:
         json.loads(output)
         score = 1.0  # 有效 JSON
     except json.JSONDecodeError:
         score = 0.0  # 无效 JSON
         
     return {"key": "json_validity", "score": score}
 
 @run_evaluator
 def length_compliance_evaluator(run, example) -> dict:
     """检查响应长度是否在期望范围内"""
     
     output = run.outputs.get("output", "")
     expected_min = example.outputs.get("min_length", 50)
     expected_max = example.outputs.get("max_length", 500)
     
     actual_length = len(output)
     
     if expected_min <= actual_length <= expected_max:
         score = 1.0
     else:
         # 基于与目标范围距离的部分分数
         if actual_length < expected_min:
             score = actual_length / expected_min
         else:  # actual_length > expected_max
             score = max(0, 1 - (actual_length - expected_max) / expected_max)
             
     return {"key": "length_compliance", "score": score}
 
 @run_evaluator
 def keyword_presence_evaluator(run, example) -> dict:
     """检查响应中是否存在所需关键词"""
     
     output = run.outputs.get("output", "").lower()
     required_keywords = example.outputs.get("required_keywords", [])
     
     if not required_keywords:
         return {"key": "keyword_presence", "score": 1.0}
         
     present_keywords = sum(1 for keyword in required_keywords if keyword.lower() in output)
     score = present_keywords / len(required_keywords)
     
     return {"key": "keyword_presence", "score": score}
 
 @run_evaluator
 def format_compliance_evaluator(run, example) -> dict:
     """检查输出是否遵循指定格式"""
     
     output = run.outputs.get("output", "")
     expected_format = example.outputs.get("expected_format", "text")
     
     score = 0.0
     
     if expected_format == "email":
         # 检查电子邮件格式
         if re.search(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', output):
             score = 1.0
     elif expected_format == "phone":
         # 检查电话号码格式
         if re.search(r'\b\d{3}-\d{3}-\d{4}\b|\b\(\d{3}\)\s*\d{3}-\d{4}\b', output):
             score = 1.0
     elif expected_format == "url":
         # 检查 URL 格式
         if re.search(r'https?://[^\s]+', output):
             score = 1.0
     elif expected_format == "markdown":
         # 检查基本 markdown 元素
         markdown_elements = ['#', '*', '**', '`', '[', ']', '(', ')']
         if any(element in output for element in markdown_elements):
             score = 1.0
     else:
         score = 1.0  # 默认为文本格式
         
     return {"key": "format_compliance", "score": score}

 def create_algorithmic_evaluation_dataset():
     """创建带有算法反馈标准的数据集"""
     
     dataset_name = "Algorithmic Feedback Evaluation"
     
     examples = [
         {
             "inputs": {
                 "task": "Generate a JSON response with user data",
                 "prompt": "创建一个包含姓名、电子邮件和电话号码的用户配置文件 JSON"
             },
             "outputs": {
                 "expected_format": "json",
                 "required_keywords": ["name", "email", "phone"],
                 "min_length": 50,
                 "max_length": 200
             }
         },
         {
             "inputs": {
                 "task": "Write a professional email response",
                 "prompt": "写一封专业的电子邮件回复客户询问"
             },
             "outputs": {
                 "expected_format": "email",
                 "required_keywords": ["dear", "thank", "regards"],
                 "min_length": 100,
                 "max_length": 300
             }
         }
     ]
     
     dataset = client.create_dataset(dataset_name=dataset_name)
     client.create_examples(
         inputs=[e["inputs"] for e in examples],
         outputs=[e["outputs"] for e in examples],
         dataset_id=dataset.id,
     )
     
     return dataset_name

 def run_algorithmic_evaluation():
     """运行带有多个算法反馈评估器的评估"""
     
     dataset_name = create_algorithmic_evaluation_dataset()
     
     # 定义被测试系统
     def test_system(inputs: dict) -> dict:
         prompt = inputs["prompt"]
         llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.3)
         response = llm.invoke(prompt)
         return {"output": response.content}
     
     # 配置所有算法评估器
     algorithmic_evaluators = [
         response_time_evaluator,
         json_validity_evaluator,
         length_compliance_evaluator,
         keyword_presence_evaluator,
         format_compliance_evaluator
     ]
     
     eval_config = RunEvalConfig(
         custom_evaluators=algorithmic_evaluators
     )
     
     # 运行评估
     results = client.run_on_dataset(
         dataset_name=dataset_name,
         llm_or_chain_factory=test_system,
         evaluation=eval_config,
         verbose=True,
         project_metadata={"evaluation_type": "algorithmic_feedback"},
     )
     
     return results

 def generate_algorithmic_feedback_report():
     """生成算法反馈指标的详细报告"""
     
     # 获取最近的评估运行
     runs = client.list_runs(project_name="your-project-name", limit=100)
     
     metrics = {
         "response_time": [],
         "json_validity": [],
         "length_compliance": [],
         "keyword_presence": [],
         "format_compliance": []
     }
     
     for run in runs:
         feedback = client.list_feedback(run_ids=[run.id])
         for f in feedback:
             if f.key in metrics:
                 metrics[f.key].append(f.score)
     
     # 计算汇总统计
     report = {}
     for metric, scores in metrics.items():
         if scores:
             report[metric] = {
                 "average": sum(scores) / len(scores),
                 "min": min(scores),
                 "max": max(scores),
                 "count": len(scores),
                 "pass_rate": sum(1 for s in scores if s >= 0.8) / len(scores)
             }
     
     print("算法反馈报告")
     print("=" * 40)
     for metric, stats in report.items():
         print(f"\n{metric.replace('_', ' ').title()}:")
         print(f"  平均分数: {stats['average']:.3f}")
         print(f"  通过率 (≥0.8): {stats['pass_rate']:.1%}")
         print(f"  范围: {stats['min']:.3f} - {stats['max']:.3f}")
     
     return report

 def benchmark_system_performance():
     """对系统性能进行基准测试"""
     
     test_queries = [
         "生成一个简单的 JSON 对象",
         "写一个长度为 100 字的段落",
         "创建一个格式化的电子邮件"
     ] * 10  # 重复以获得更好的统计
     
     response_times = []
     
     for query in test_queries:
         start_time = time.time()
         
         # 运行系统
         llm = ChatOpenAI(model="gpt-3.5-turbo")
         response = llm.invoke(query)
         
         end_time = time.time()
         response_times.append(end_time - start_time)
     
     # 计算性能指标
     avg_time = sum(response_times) / len(response_times)
     p95_time = sorted(response_times)[int(0.95 * len(response_times))]