-
Notifications
You must be signed in to change notification settings - Fork 342
Expand file tree
/
Copy pathexpected_output.py
More file actions
111 lines (86 loc) · 3.81 KB
/
Copy pathexpected_output.py
File metadata and controls
111 lines (86 loc) · 3.81 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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
"""
LangGraph definition for research-assistant v1.0.0
Generated by gitagent export --format langgraph
"""
from __future__ import annotations
from typing import Annotated, TypedDict
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage
from langchain_core.tools import tool
from langchain.chat_models import init_chat_model
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
# Agent metadata
AGENT_NAME = "research-assistant"
AGENT_VERSION = "1.0.0"
AGENT_DESCRIPTION = "Two-step research assistant — searches the web, then summarizes findings"
MODEL = "claude-sonnet-4-5"
RECURSION_LIMIT = 25
# System prompt (SOUL.md + RULES.md + DUTIES.md + compliance)
SYSTEM_PROMPT = """# research-assistant
Two-step research assistant — searches the web, then summarizes findings
# Research Assistant
You are a research assistant. Find authoritative sources, extract the relevant
facts, and summarize them honestly. Cite every claim. When uncertain, say so.
"""
# Skill instructions
SKILL_SUMMARIZE_INSTRUCTIONS = """Condense the gathered sources into a faithful, cited summary
Produce a concise summary of the gathered material. Every factual claim must
be attributed to a specific source URL. If the sources disagree, say so
explicitly rather than picking a side."""
SKILL_WEB_RESEARCH_INSTRUCTIONS = """Search the web for sources relevant to the user's question
Allowed tools: web-search
Use the `web-search` tool to find authoritative sources. Prefer primary sources
(official docs, academic papers) over secondary commentary. Capture the URL of
every source consulted."""
# Graph state
class AgentState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
# Tools (from tools/*.yaml)
@tool
def web_search(query: str) -> str:
"""Search the public web for a query"""
raise NotImplementedError("Implement tool: web-search")
TOOLS = [web_search]
tool_node = ToolNode(TOOLS)
llm = init_chat_model(MODEL)
llm_with_tools = llm.bind_tools(TOOLS) if TOOLS else llm
# Hooks (pre_tool_use → before_tool callback)
def before_tool(state: AgentState) -> AgentState:
"""No pre_tool_use hooks configured in hooks/hooks.yaml."""
return state
# NOTE: post_tool_use, on_session_start, etc. have no LangGraph equivalent — skipped.
# NOTE: memory/ has no direct LangGraph equivalent — implement via a checkpointer.
# Skill nodes (one per skills/<name>/SKILL.md)
def skill_summarize(state: AgentState) -> dict:
"""Skill node: summarize — Condense the gathered sources into a faithful, cited summary"""
prompt = SYSTEM_PROMPT + "\n\n" + SKILL_SUMMARIZE_INSTRUCTIONS
messages = [SystemMessage(content=prompt), *state["messages"]]
response = llm_with_tools.invoke(messages)
return {"messages": [response]}
def skill_web_research(state: AgentState) -> dict:
"""Skill node: web-research — Search the web for sources relevant to the user's question"""
prompt = SYSTEM_PROMPT + "\n\n" + SKILL_WEB_RESEARCH_INSTRUCTIONS
messages = [SystemMessage(content=prompt), *state["messages"]]
response = llm_with_tools.invoke(messages)
return {"messages": [response]}
# Graph construction
graph = StateGraph(AgentState)
graph.add_node("tools", tool_node)
# Skillflow: research
graph.add_node("research", skill_web_research)
graph.add_node("summarize", skill_summarize)
graph.add_edge(START, "research")
graph.add_edge("research", "summarize")
graph.add_edge("summarize", END)
# Compile
app = graph.compile()
if __name__ == "__main__":
import os
user_input = os.environ.get("GITAGENT_PROMPT", "Hello")
result = app.invoke(
{"messages": [HumanMessage(content=user_input)]},
config={"recursion_limit": RECURSION_LIMIT},
)
for message in result["messages"]:
print(message)