这里个大家介绍一个简单易用的智能体——测试用例生成器，它是一个完整的 LangGraph 智能体，这个智能体会包含：感知、规划、生成、反思四个核心能力。

完整智能体代码

废话不多说，直接上代码：

1import os
2import json
3import sqlite3
4import pandas as pd
5from datetime import datetime
6from typing import List, Dict, TypedDict, Annotated
7from pydantic import BaseModel, Field, field_validator
8import operator
9
10# ====== LangChain & LangGraph 导入 ======
11from langchain_core.prompts import PromptTemplate
12from langchain_core.output_parsers import PydanticOutputParser
13from langchain_core.runnables import RunnableLambda
14from langchain_community.chat_models import ChatTongyi
15from langchain_community.vectorstores import Chroma
16from langchain_community.embeddings import DashScopeEmbeddings
17from langchain_core.messages import ChatMessageHistory
18
19# ====== LangGraph 导入 ======
20from langgraph.graph import StateGraph
21from langgraph.graph.message import add_messages
22
23# ====== 1. 定义测试用例结构（保持原有）======
24class TestCase(BaseModel):
25    description: str = Field(description="测试用例描述")
26    steps: List[str] = Field(description="操作步骤列表，每步为字符串")
27    expectedResult: str = Field(description="预期结果")
28    priority: str = Field(description="优先级")
29
30@field_validator("priority")
31def validate_priority(cls, v):
32    priority_map = {
33        "high": "High", "medium": "Medium", "low": "Low",
34        "高": "High", "中": "Medium", "低": "Low",
35        "high优先级": "High", "medium优先级": "Medium", "low优先级": "Low"
36    }
37    normalized_v = v.lower() if isinstance(v, str) else str(v).lower()
38    if normalized_v in priority_map:
39        return priority_map[normalized_v]
40    elif normalized_v.title() in {"High", "Medium", "Low"}:
41        return normalized_v.title()
42    else:
43        raise ValueError('priority 必须是 "High"、"Medium" 或 "Low"')
44
45# ====== 2. 智能体状态定义 ======
46class AgentState(TypedDict):
47    requirement: str                    # 输入需求
48    test_cases: List[Dict]              # 生成的测试用例
49    memory: List[str]                   # 工作记忆
50    feedback: str                       # 质量反馈
51    quality_score: float                # 质量评分
52    reflection_notes: List[str]         # 反思笔记
53    execution_history: List[str]        # 执行历史
54
55# ====== 3. 初始化 LLM 和向量数据库 ======
56DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY") or "sk-7"
57
58llm = ChatTongyi(
59    model_name="qwen-max",
60    dashscope_api_key=DASHSCOPE_API_KEY,
61    temperature=0.3,
62    streaming=False,
63)
64
65# 初始化向量数据库用于记忆
66memory_db = Chroma(
67    collection_name="test_case_memory",
68    embedding_function=DashScopeEmbeddings(dashscope_api_key=DASHSCOPE_API_KEY)
69)
70
71parser = PydanticOutputParser(pydantic_object=TestCase)
72
73# ====== 4. 智能体各节点实现 ======
74
75def perception_node(state: AgentState):
76    """感知节点：分析需求并从历史中检索相关信息"""
77    print("[感知节点] 分析需求并检索历史信息...")
78    
79    # 检索相关的历史测试用例
80    similar_docs = memory_db.similarity_search(state["requirement"], k=3)
81    
82    # 构建上下文
83    context = f"相关历史用例: {[doc.page_content[:100] for doc in similar_docs]}"
84    
85    updated_state = state.copy()
86    updated_state["memory"] = [context]
87    updated_state["execution_history"].append("感知: 检索历史相关用例")
88    
89    return updated_state
90
91def planning_node(state: AgentState):
92    """规划节点：制定生成策略"""
93    print("[规划节点] 制定测试用例生成策略...")
94    
95    planning_prompt = f"""
96    为以下需求制定测试用例生成策略：
97    需求: {state["requirement"]}
98    
99    请提供:
100    1. 测试类型分布（正向/负向/边界值）
101    2. 优先级分布建议
102    3. 关键测试点
103    """
104    
105    strategy = llm.invoke(planning_prompt).content
106    
107    updated_state = state.copy()
108    updated_state["memory"].append(f"生成策略: {strategy}")
109    updated_state["execution_history"].append("规划: 制定生成策略")
110    
111    return updated_state
112
113def generation_node(state: AgentState):
114    """生成节点：核心测试用例生成（基于你原有逻辑）"""
115    print("[生成节点] 正在生成测试用例...")
116    
117    prompt_template = """
118    你是一个资深测试工程师，请为以下功能生成一个高质量测试用例。
119    {format_instructions}
120    
121    历史参考: {context}
122    
123    功能描述：{function_desc}
124    
125    请只输出一个测试用例，不要任何其他文字。
126    """
127    
128    context = "\n".join(state.get("memory", []))
129    
130    prompt = PromptTemplate(
131        template=prompt_template,
132        input_variables=["function_desc", "context"],
133        partial_variables={"format_instructions": parser.get_format_instructions()},
134    )
135    
136    chain = prompt | llm | parser
137    
138    # 生成多个测试用例
139    test_cases = []
140    for i in range(3):  # 生成3个用例
141        try:
142            result = chain.invoke({
143                "function_desc": state["requirement"],
144                "context": context
145            })
146            test_cases.append(result.dict())
147            print(f"用例 {i+1} 生成成功")
148        except Exception as e:
149            print(f"用例 {i+1} 生成失败: {str(e)}")
150    
151    updated_state = state.copy()
152    updated_state["test_cases"] = test_cases
153    updated_state["execution_history"].append(f"生成: 创建了 {len(test_cases)} 个用例")
154    
155    return updated_state
156
157def reflection_node(state: AgentState):
158    """反思节点：评估生成质量并提供改进建议"""
159    print("[反思节点] 评估测试用例质量...")
160    
161    if not state["test_cases"]:
162        updated_state = state.copy()
163        updated_state["quality_score"] = 0.0
164        updated_state["feedback"] = "未生成任何测试用例"
165        updated_state["reflection_notes"] = ["需要重新分析需求"]
166        updated_state["execution_history"].append("反思: 无用例可评估")
167        return updated_state
168    
169    # 评估第一个用例的质量
170    test_case = state["test_cases"][0]
171    evaluation_prompt = f"""
172    评估以下测试用例的质量（1-10分）：
173    需求: {state["requirement"]}
174    用例: {test_case}
175    
176    请提供:
177    1. 整体质量评分（1-10）
178    2. 相关性评分（1-10）
179    3. 可执行性评分（1-10）
180    4. 改进建议
181    5. 是否满足需求
182    """
183    
184    evaluation = llm.invoke(evaluation_prompt).content
185    
186    # 解析评分（简化处理）
187    try:
188        score = float([line for line in evaluation.split('\n') if '整体质量评分' in line][0][-2:])
189    except:
190        score = 7.0  # 默认分数
191    
192    updated_state = state.copy()
193    updated_state["quality_score"] = score
194    updated_state["feedback"] = evaluation
195    updated_state["reflection_notes"] = [f"质量评分: {score}/10", "评估完成"]
196    updated_state["execution_history"].append(f"反思: 质量评分为 {score}/10")
197    
198    return updated_state
199
200def learning_node(state: AgentState):
201    """学习节点：将经验存入长期记忆"""
202    print("[学习节点] 存储经验到长期记忆...")
203    
204    # 将生成的用例和评估结果存入向量数据库
205    for i, case in enumerate(state["test_cases"][:2]):  # 存储前2个
206        memory_text = f"需求: {state['requirement']} | 用例: {case} | 评估: {state['feedback'][:200]}"
207        memory_db.add_texts(
208            texts=[memory_text],
209            metadatas=[{
210                "requirement": state["requirement"],
211                "quality_score": state["quality_score"],
212                "timestamp": datetime.now().isoformat()
213            }]
214        )
215    
216    updated_state = state.copy()
217    updated_state["execution_history"].append("学习: 存储生成经验")
218    
219    return updated_state
220
221# ====== 5. 构建 LangGraph 智能体 ======
222def create_test_case_agent():
223    """创建完整的测试用例生成智能体"""
224    workflow = StateGraph(AgentState)
225    
226    # 添加节点
227    workflow.add_node("perception", perception_node)
228    workflow.add_node("planning", planning_node)
229    workflow.add_node("generation", generation_node)
230    workflow.add_node("reflection", reflection_node)
231    workflow.add_node("learning", learning_node)
232    
233    # 定义流转
234    workflow.set_entry_point("perception")
235    workflow.add_edge("perception", "planning")
236    workflow.add_edge("planning", "generation")
237    workflow.add_edge("generation", "reflection")
238    workflow.add_edge("reflection", "learning")
239    
240    # 智能体结束
241    graph = workflow.compile()
242    return graph
243
244# ====== 6. 辅助函数 ======
245def save_to_db(test_cases: List[Dict], function_desc: str):
246    """保存到SQLite数据库"""
247    conn = sqlite3.connect("test_cases.db")
248    conn.execute("""
249        CREATE TABLE IF NOT EXISTS test_cases (
250            id INTEGER PRIMARY KEY AUTOINCREMENT,
251            function TEXT,
252            test_cases TEXT,
253            timestamp TEXT
254        )
255    """)
256    conn.execute(
257        "INSERT INTO test_cases (function, test_cases, timestamp) VALUES (?, ?, ?)",
258        (function_desc, json.dumps(test_cases, ensure_ascii=False), datetime.now().isoformat()),
259    )
260    conn.commit()
261    conn.close()
262    print("用例已保存到数据库")
263
264def export_to_csv(test_cases: List[Dict]):
265    """导出为CSV文件"""
266    for tc in test_cases:
267        tc["steps"] = "; ".join(tc["steps"])
268    df = pd.DataFrame(test_cases)
269    df.to_csv("test_cases.csv", index=False, encoding="utf-8-sig")
270    print("已导出到 test_cases.csv")
271
272# ====== 7. 主程序 ======
273def run_intelligent_test_case_agent(requirement: str, num_cases: int = 3):
274    """运行智能体生成测试用例"""
275    print(f"启动测试用例生成智能体...")
276    print(f"需求: {requirement}")
277    
278    # 初始化智能体
279    agent = create_test_case_agent()
280    
281    # 初始化状态
282    initial_state = {
283        "requirement": requirement,
284        "test_cases": [],
285        "memory": [],
286        "feedback": "",
287        "quality_score": 0.0,
288        "reflection_notes": [],
289        "execution_history": ["开始执行"]
290    }
291    
292    # 执行智能体
293    final_state = agent.invoke(initial_state)
294    
295    # 输出结果
296    print("\n" + "="*60)
297    print("智能体执行结果:")
298    print("="*60)
299    print(f"质量评分: {final_state['quality_score']}/10")
300    print(f"反思笔记: {final_state['reflection_notes']}")
301    print(f"执行历史: {' -> '.join(final_state['execution_history'])}")
302    
303    if final_state["test_cases"]:
304        print(f"\n生成了 {len(final_state['test_cases'])} 个测试用例:")
305        for i, tc in enumerate(final_state["test_cases"]):
306            print(f"\n用例 {i+1}: {tc['description']}")
307            print(f"优先级: {tc['priority']}")
308            print(f"步骤: {' -> '.join(tc['steps'][:2])}...")
309        
310        # 保存结果
311        save_to_db(final_state["test_cases"], requirement)
312        export_to_csv(final_state["test_cases"].copy())  # 保存副本
313        
314        print(f"\n智能体任务完成！")
315    else:
316        print("\n未能生成有效测试用例")
317    
318    return final_state
319
320if __name__ == "__main__":
321    import argparse
322    
323    parser = argparse.ArgumentParser(description="AI 测试用例生成智能体")
324    parser.add_argument("function", help="功能描述，例如 '用户登录功能'")
325    parser.add_argument("--num", type=int, default=3, help="生成用例数量（默认3）")
326    args = parser.parse_args()
327    
328    # 运行智能体
329    result = run_intelligent_test_case_agent(args.function, args.num)
330

运行说明

1. 安装依赖

pip install langgraph langchain-community dashscope chromadb pandas pydantic

2. 设置 API Key

export DASHSCOPE_API_KEY="your_actual_key"  # Linux/Mac
set DASHSCOPE_API_KEY="your_actual_key"     # Windows

3. 运行智能体

python your_file.py "用户登录功能"

 智能体架构解析

核心节点：

1. 感知节点: 从向量数据库检索历史用例，为生成提供上下文
2. 规划节点: 制定生成策略，决定测试类型分布
3. 生成节点: 基于你的原有逻辑生成测试用例
4. 反思节点: 评估生成质量，提供改进建议
5. 学习节点: 将经验存入长期记忆

智能体现在：

• 自主规划: 不是简单执行，而是先分析再生成
• 持续学习: 每次生成都会存入记忆，供下次使用
• 质量控制: 自动生成质量评估和改进建议
• 环境感知: 能利用历史数据提升生成质量

智能体 vs 工具对比

这个智能体展示了 "感知 → 规划 → 执行 → 反思 → 学习" 的完整闭环，真正体现了智能体的自主性和适应性！