-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmain.py
More file actions
75 lines (60 loc) · 1.96 KB
/
Copy pathmain.py
File metadata and controls
75 lines (60 loc) · 1.96 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
"""
Entry point for the Self-Validating Agent Graph.
Usage:
python main.py
python main.py --query "Explain what LangGraph is"
python main.py --strategy rule_based --max-retries 3
"""
import argparse
import os
import sys
from graph.graph_builder import build_graph
def run(query: str, max_retries: int = 2, strategy: str = "llm_judge") -> None:
os.environ["SCORING_STRATEGY"] = strategy
initial_state = {
"query": query,
"conversation": [],
"current_agent": "",
"current_agent_output": None,
"validation_score": None,
"validation_feedback": None,
"retry_count": 0,
"max_retries": max_retries,
"halted": False,
"halt_reason": None,
"final_answer": None,
}
print(f"\n{'='*60}")
print(f"Query: {query}")
print(f"Strategy: {strategy} | Max retries: {max_retries}")
print(f"{'='*60}\n")
graph = build_graph()
final_state = graph.invoke(initial_state)
print(f"\n{'='*60}")
if final_state.get("halted"):
print("RESULT: HALTED")
print(f"Reason: {final_state.get('halt_reason')}")
else:
print("RESULT: SUCCESS")
print(f"Final answer:\n{final_state.get('final_answer')}")
print(f"{'='*60}\n")
def main():
parser = argparse.ArgumentParser(description="Self-Validating Agent Graph")
parser.add_argument(
"--query",
type=str,
default="What is LangGraph and how does it relate to LangChain?",
)
parser.add_argument(
"--strategy",
choices=["rule_based", "llm_judge"],
default="llm_judge",
)
parser.add_argument("--max-retries", type=int, default=2)
args = parser.parse_args()
if not os.environ.get("ANTHROPIC_API_KEY"):
print("Error: ANTHROPIC_API_KEY environment variable is not set.", file=sys.stderr)
sys.exit(1)
run(query=args.query, max_retries=args.max_retries, strategy=args.strategy)
if __name__ == "__main__":
main()