"""
Graph Retrieval Tools Service
Encapsulates graph search, node retrieval, edge queries, and other tools for use by Report Agent.
Replaces zep_tools.py — all Zep Cloud calls replaced by GraphStorage.
Core Retrieval Tools (Optimized):
1. InsightForge (Deep Insight Retrieval) - Most powerful hybrid search, automatically generates sub-questions and multi-dimensional retrieval
2. PanoramaSearch (Breadth Search) - Get comprehensive view, including expired content
3. QuickSearch (Simple Search) - Quick retrieval
"""
import json
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from ..utils.logger import get_logger
from ..utils.llm_client import LLMClient
from ..storage import GraphStorage
logger = get_logger('mirofish.graph_tools')
@dataclass
class SearchResult:
"""Search Result"""
facts: List[str]
edges: List[Dict[str, Any]]
nodes: List[Dict[str, Any]]
query: str
total_count: int
def to_dict(self) -> Dict[str, Any]:
return {
"facts": self.facts,
"edges": self.edges,
"nodes": self.nodes,
"query": self.query,
"total_count": self.total_count
}
def to_text(self) -> str:
"""Convert to text format for LLM understanding"""
text_parts = [f"Search Query: {self.query}", f"Found {self.total_count} related results"]
if self.facts:
text_parts.append("\n### Related Facts:")
for i, fact in enumerate(self.facts, 1):
text_parts.append(f"{i}. {fact}")
return "\n".join(text_parts)
@dataclass
class NodeInfo:
"""Node Information"""
uuid: str
name: str
labels: List[str]
summary: str
attributes: Dict[str, Any]
def to_dict(self) -> Dict[str, Any]:
return {
"uuid": self.uuid,
"name": self.name,
"labels": self.labels,
"summary": self.summary,
"attributes": self.attributes
}
def to_text(self) -> str:
"""Convert to text format"""
entity_type = next((la for la in self.labels if la not in ["Entity", "Node"]), "Unknown type")
return f"Entity: {self.name} (Type: {entity_type})\nSummary: {self.summary}"
@dataclass
class EdgeInfo:
"""Edge Information"""
uuid: str
name: str
fact: str
source_node_uuid: str
target_node_uuid: str
source_node_name: Optional[str] = None
target_node_name: Optional[str] = None
# Temporal information (may be absent in Neo4j — kept for interface compat)
created_at: Optional[str] = None
valid_at: Optional[str] = None
invalid_at: Optional[str] = None
expired_at: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
return {
"uuid": self.uuid,
"name": self.name,
"fact": self.fact,
"source_node_uuid": self.source_node_uuid,
"target_node_uuid": self.target_node_uuid,
"source_node_name": self.source_node_name,
"target_node_name": self.target_node_name,
"created_at": self.created_at,
"valid_at": self.valid_at,
"invalid_at": self.invalid_at,
"expired_at": self.expired_at
}
def to_text(self, include_temporal: bool = False) -> str:
"""Convert to text format"""
source = self.source_node_name or self.source_node_uuid[:8]
target = self.target_node_name or self.target_node_uuid[:8]
base_text = f"Relationship: {source} --[{self.name}]--> {target}\nFact: {self.fact}"
if include_temporal:
valid_at = self.valid_at or "Unknown"
invalid_at = self.invalid_at or "Present"
base_text += f"\nTime Range: {valid_at} - {invalid_at}"
if self.expired_at:
base_text += f" (Expired: {self.expired_at})"
return base_text
@property
def is_expired(self) -> bool:
"""Whether already expired"""
return self.expired_at is not None
@property
def is_invalid(self) -> bool:
"""Whether already invalid"""
return self.invalid_at is not None
@dataclass
class InsightForgeResult:
"""
Deep Insight Retrieval Result (InsightForge)
Contains retrieval results from multiple sub-questions and integrated analysis
"""
query: str
simulation_requirement: str
sub_queries: List[str]
# Retrieval results by dimension
semantic_facts: List[str] = field(default_factory=list)
entity_insights: List[Dict[str, Any]] = field(default_factory=list)
relationship_chains: List[str] = field(default_factory=list)
# Statistical information
total_facts: int = 0
total_entities: int = 0
total_relationships: int = 0
def to_dict(self) -> Dict[str, Any]:
return {
"query": self.query,
"simulation_requirement": self.simulation_requirement,
"sub_queries": self.sub_queries,
"semantic_facts": self.semantic_facts,
"entity_insights": self.entity_insights,
"relationship_chains": self.relationship_chains,
"total_facts": self.total_facts,
"total_entities": self.total_entities,
"total_relationships": self.total_relationships
}
def to_text(self) -> str:
"""Convert to detailed text format for LLM understanding"""
text_parts = [
f"## Future Prediction Deep Analysis",
f"Analysis Query: {self.query}",
f"Prediction Scenario: {self.simulation_requirement}",
f"\n### Prediction Data Statistics",
f"- Related Prediction Facts: {self.total_facts}",
f"- Involved Entities: {self.total_entities}",
f"- Relationship Chains: {self.total_relationships}"
]
if self.sub_queries:
text_parts.append(f"\n### Analysis Sub-Questions")
for i, sq in enumerate(self.sub_queries, 1):
text_parts.append(f"{i}. {sq}")
if self.semantic_facts:
text_parts.append(f"\n### Key Facts (Please quote these verbatim in the report)")
for i, fact in enumerate(self.semantic_facts, 1):
text_parts.append(f'{i}. "{fact}"')
if self.entity_insights:
text_parts.append(f"\n### Core Entities")
for entity in self.entity_insights:
text_parts.append(f"- **{entity.get('name', 'Unknown')}** ({entity.get('type', 'Entity')})")
if entity.get('summary'):
text_parts.append(f" Summary: \"{entity.get('summary')}\"")
if entity.get('related_facts'):
text_parts.append(f" Related Facts: {len(entity.get('related_facts', []))} facts")
if self.relationship_chains:
text_parts.append(f"\n### Relationship Chains")
for chain in self.relationship_chains:
text_parts.append(f"- {chain}")
return "\n".join(text_parts)
@dataclass
class PanoramaResult:
"""
Breadth Search Result (Panorama)
Contains all related information, including expired content
"""
query: str
all_nodes: List[NodeInfo] = field(default_factory=list)
all_edges: List[EdgeInfo] = field(default_factory=list)
active_facts: List[str] = field(default_factory=list)
historical_facts: List[str] = field(default_factory=list)
total_nodes: int = 0
total_edges: int = 0
active_count: int = 0
historical_count: int = 0
def to_dict(self) -> Dict[str, Any]:
return {
"query": self.query,
"all_nodes": [n.to_dict() for n in self.all_nodes],
"all_edges": [e.to_dict() for e in self.all_edges],
"active_facts": self.active_facts,
"historical_facts": self.historical_facts,
"total_nodes": self.total_nodes,
"total_edges": self.total_edges,
"active_count": self.active_count,
"historical_count": self.historical_count
}
def to_text(self) -> str:
"""Convert to text format (complete version, no truncation)"""
text_parts = [
f"## Breadth Search Results (Future Panoramic View)",
f"Query: {self.query}",
f"\n### Statistics",
f"- Total Nodes: {self.total_nodes}",
f"- Total Edges: {self.total_edges}",
f"- Current Valid Facts: {self.active_count}",
f"- Historical/Expired Facts: {self.historical_count}"
]
if self.active_facts:
text_parts.append(f"\n### Current Valid Facts (Simulation Results Verbatim)")
for i, fact in enumerate(self.active_facts, 1):
text_parts.append(f'{i}. "{fact}"')
if self.historical_facts:
text_parts.append(f"\n### Historical/Expired Facts (Evolution Record)")
for i, fact in enumerate(self.historical_facts, 1):
text_parts.append(f'{i}. "{fact}"')
if self.all_nodes:
text_parts.append(f"\n### Involved Entities")
for node in self.all_nodes:
entity_type = next((la for la in node.labels if la not in ["Entity", "Node"]), "Entity")
text_parts.append(f"- **{node.name}** ({entity_type})")
return "\n".join(text_parts)
@dataclass
class AgentInterview:
"""Single Agent Interview Result"""
agent_name: str
agent_role: str
agent_bio: str
question: str
response: str
key_quotes: List[str] = field(default_factory=list)
def to_dict(self) -> Dict[str, Any]:
return {
"agent_name": self.agent_name,
"agent_role": self.agent_role,
"agent_bio": self.agent_bio,
"question": self.question,
"response": self.response,
"key_quotes": self.key_quotes
}
def to_text(self) -> str:
text = f"**{self.agent_name}** ({self.agent_role})\n"
text += f"_Bio: {self.agent_bio}_\n\n"
text += f"**Q:** {self.question}\n\n"
text += f"**A:** {self.response}\n"
if self.key_quotes:
text += "\n**Key Quotes:**\n"
for quote in self.key_quotes:
clean_quote = quote.replace('\u201c', '').replace('\u201d', '').replace('"', '')
clean_quote = clean_quote.replace('\u300c', '').replace('\u300d', '')
clean_quote = clean_quote.strip()
while clean_quote and clean_quote[0] in '，,；;：:、。！？\n\r\t ':
clean_quote = clean_quote[1:]
skip = False
for d in '123456789':
if f'\u95ee\u9898{d}' in clean_quote:
skip = True
break
if skip:
continue
if len(clean_quote) > 150:
dot_pos = clean_quote.find('\u3002', 80)
if dot_pos > 0:
clean_quote = clean_quote[:dot_pos + 1]
else:
clean_quote = clean_quote[:147] + "..."
if clean_quote and len(clean_quote) >= 10:
text += f'> "{clean_quote}"\n'
return text
@dataclass
class InterviewResult:
"""
Interview Result
Contains interview responses from multiple simulated Agents
"""
interview_topic: str
interview_questions: List[str]
selected_agents: List[Dict[str, Any]] = field(default_factory=list)
interviews: List[AgentInterview] = field(default_factory=list)
selection_reasoning: str = ""
summary: str = ""
total_agents: int = 0
interviewed_count: int = 0
def to_dict(self) -> Dict[str, Any]:
return {
"interview_topic": self.interview_topic,
"interview_questions": self.interview_questions,
"selected_agents": self.selected_agents,
"interviews": [i.to_dict() for i in self.interviews],
"selection_reasoning": self.selection_reasoning,
"summary": self.summary,
"total_agents": self.total_agents,
"interviewed_count": self.interviewed_count
}
def to_text(self) -> str:
"""Convert to detailed text format for LLM understanding and report reference"""
text_parts = [
"## Deep Interview Report",
f"**Interview Topic:** {self.interview_topic}",
f"**Interviewees:** {self.interviewed_count} / {self.total_agents} Simulated Agents",
"\n### Selection Rationale",
self.selection_reasoning or "(Automatic Selection)",
"\n---",
"\n### Interview Transcripts",
]
if self.interviews:
for i, interview in enumerate(self.interviews, 1):
text_parts.append(f"\n#### Interview #{i}: {interview.agent_name}")
text_parts.append(interview.to_text())
text_parts.append("\n---")
else:
text_parts.append("(No interview records)\n\n---")
text_parts.append("\n### Interview Summary & Key Insights")
text_parts.append(self.summary or "(No summary)")
return "\n".join(text_parts)
class GraphToolsService:
"""
Graph Retrieval Tools Service (via GraphStorage / Neo4j)
[Core Retrieval Tools - Optimized]
1. insight_forge - Deep Insight Retrieval (Most powerful, auto-generates sub-questions, multi-dimensional retrieval)
2. panorama_search - Breadth Search (Get comprehensive view, including expired content)
3. quick_search - Simple Search (Quick retrieval)
4. interview_agents - Deep Interview (Interview simulated Agents, obtain multi-perspective insights)
[Basic Tools]
- search_graph - Graph semantic search
- get_all_nodes - Get all nodes in graph
- get_all_edges - Get all edges in graph (with temporal information)
- get_node_detail - Get detailed node information
- get_node_edges - Get edges related to a node
- get_entities_by_type - Get entities by type
- get_entity_summary - Get entity relationship summary
"""
def __init__(self, storage: GraphStorage, llm_client: Optional[LLMClient] = None):
self.storage = storage
self._llm_client = llm_client
logger.info("GraphToolsService initialization complete")
@property
def llm(self) -> LLMClient:
"""Lazy initialization of LLM client"""
if self._llm_client is None:
self._llm_client = LLMClient()
return self._llm_client
# ========== Basic Tools ==========
def search_graph(
self,
graph_id: str,
query: str,
limit: int = 10,
scope: str = "edges"
) -> SearchResult:
"""
Graph semantic search (hybrid: vector + BM25 via Neo4j)
Args:
graph_id: Graph ID
query: Search query
limit: Number of results to return
scope: Search scope, "edges" or "nodes" or "both"
Returns:
SearchResult
"""
logger.info(f"Graph search: graph_id={graph_id}, query={query[:50]}...")
try:
search_results = self.storage.search(
graph_id=graph_id,
query=query,
limit=limit,
scope=scope,
)
facts = []
edges = []
nodes = []
# Parse edge results
if hasattr(search_results, 'edges'):
edge_list = search_results.edges
elif isinstance(search_results, dict) and 'edges' in search_results:
edge_list = search_results['edges']
else:
edge_list = []
for edge in edge_list:
if isinstance(edge, dict):
fact = edge.get('fact', '')
if fact:
facts.append(fact)
edges.append({
"uuid": edge.get('uuid', ''),
"name": edge.get('name', ''),
"fact": fact,
"source_node_uuid": edge.get('source_node_uuid', ''),
"target_node_uuid": edge.get('target_node_uuid', ''),
})
# Parse node results
if hasattr(search_results, 'nodes'):
node_list = search_results.nodes
elif isinstance(search_results, dict) and 'nodes' in search_results:
node_list = search_results['nodes']
else:
node_list = []
for node in node_list:
if isinstance(node, dict):
nodes.append({
"uuid": node.get('uuid', ''),
"name": node.get('name', ''),
"labels": node.get('labels', []),
"summary": node.get('summary', ''),
})
summary = node.get('summary', '')
if summary:
facts.append(f"[{node.get('name', '')}]: {summary}")
logger.info(f"Search complete: Found {len(facts)} related facts")
return SearchResult(
facts=facts,
edges=edges,
nodes=nodes,
query=query,
total_count=len(facts)
)
except Exception as e:
logger.warning(f"Graph search failed, degrading to local search: {str(e)}")
return self._local_search(graph_id, query, limit, scope)
def _local_search(
self,
graph_id: str,
query: str,
limit: int = 10,
scope: str = "edges"
) -> SearchResult:
"""
Local keyword matching search (fallback approach)
"""
logger.info(f"Using local search: query={query[:30]}...")
facts = []
edges_result = []
nodes_result = []
query_lower = query.lower()
keywords = [w.strip() for w in query_lower.replace(',', ' ').replace('，', ' ').split() if len(w.strip()) > 1]
def match_score(text: str) -> int:
if not text:
return 0
text_lower = text.lower()
if query_lower in text_lower:
return 100
score = 0
for keyword in keywords:
if keyword in text_lower:
score += 10
return score
try:
if scope in ["edges", "both"]:
all_edges = self.storage.get_all_edges(graph_id)
scored_edges = []
for edge in all_edges:
score = match_score(edge.get("fact", "")) + match_score(edge.get("name", ""))
if score > 0:
scored_edges.append((score, edge))
scored_edges.sort(key=lambda x: x[0], reverse=True)
for score, edge in scored_edges[:limit]:
fact = edge.get("fact", "")
if fact:
facts.append(fact)
edges_result.append({
"uuid": edge.get("uuid", ""),
"name": edge.get("name", ""),
"fact": fact,
"source_node_uuid": edge.get("source_node_uuid", ""),
"target_node_uuid": edge.get("target_node_uuid", ""),
})
if scope in ["nodes", "both"]:
all_nodes = self.storage.get_all_nodes(graph_id)
scored_nodes = []
for node in all_nodes:
score = match_score(node.get("name", "")) + match_score(node.get("summary", ""))
if score > 0:
scored_nodes.append((score, node))
scored_nodes.sort(key=lambda x: x[0], reverse=True)
for score, node in scored_nodes[:limit]:
nodes_result.append({
"uuid": node.get("uuid", ""),
"name": node.get("name", ""),
"labels": node.get("labels", []),
"summary": node.get("summary", ""),
})
summary = node.get("summary", "")
if summary:
facts.append(f"[{node.get('name', '')}]: {summary}")
logger.info(f"Local search complete: Found {len(facts)} related facts")
except Exception as e:
logger.error(f"Local search failed: {str(e)}")
return SearchResult(
facts=facts,
edges=edges_result,
nodes=nodes_result,
query=query,
total_count=len(facts)
)
def get_all_nodes(self, graph_id: str) -> List[NodeInfo]:
"""Get all nodes in the graph"""
logger.info(f"Getting all nodes in graph {graph_id}...")
raw_nodes = self.storage.get_all_nodes(graph_id)
result = []
for node in raw_nodes:
result.append(NodeInfo(
uuid=node.get("uuid", ""),
name=node.get("name", ""),
labels=node.get("labels", []),
summary=node.get("summary", ""),
attributes=node.get("attributes", {})
))
logger.info(f"Retrieved {len(result)} nodes")
return result
def get_all_edges(self, graph_id: str, include_temporal: bool = True) -> List[EdgeInfo]:
"""Get all edges in the graph (with temporal information)"""
logger.info(f"Getting all edges in graph {graph_id}...")
raw_edges = self.storage.get_all_edges(graph_id)
result = []
for edge in raw_edges:
edge_info = EdgeInfo(
uuid=edge.get("uuid", ""),
name=edge.get("name", ""),
fact=edge.get("fact", ""),
source_node_uuid=edge.get("source_node_uuid", ""),
target_node_uuid=edge.get("target_node_uuid", "")
)
if include_temporal:
edge_info.created_at = edge.get("created_at")
edge_info.valid_at = edge.get("valid_at")
edge_info.invalid_at = edge.get("invalid_at")
edge_info.expired_at = edge.get("expired_at")
result.append(edge_info)
logger.info(f"Retrieved {len(result)} edges")
return result
def get_node_detail(self, node_uuid: str) -> Optional[NodeInfo]:
"""Get detailed information about a single node"""
logger.info(f"Getting node details: {node_uuid[:8]}...")
try:
node = self.storage.get_node(node_uuid)
if not node:
return None
return NodeInfo(
uuid=node.get("uuid", ""),
name=node.get("name", ""),
labels=node.get("labels", []),
summary=node.get("summary", ""),
attributes=node.get("attributes", {})
)
except Exception as e:
logger.error(f"Failed to get node details: {str(e)}")
return None
def get_node_edges(self, graph_id: str, node_uuid: str) -> List[EdgeInfo]:
"""
Get all edges related to a node
Optimized: uses storage.get_node_edges() (O(degree) Cypher)
instead of loading ALL edges and filtering.
"""
logger.info(f"Getting edges related to node {node_uuid[:8]}...")
try:
raw_edges = self.storage.get_node_edges(node_uuid)
result = []
for edge in raw_edges:
result.append(EdgeInfo(
uuid=edge.get("uuid", ""),
name=edge.get("name", ""),
fact=edge.get("fact", ""),
source_node_uuid=edge.get("source_node_uuid", ""),
target_node_uuid=edge.get("target_node_uuid", ""),
created_at=edge.get("created_at"),
valid_at=edge.get("valid_at"),
invalid_at=edge.get("invalid_at"),
expired_at=edge.get("expired_at"),
))
logger.info(f"Found {len(result)} edges related to the node")
return result
except Exception as e:
logger.warning(f"Failed to get node edges: {str(e)}")
return []
def get_entities_by_type(
self,
graph_id: str,
entity_type: str
) -> List[NodeInfo]:
"""Get entities by type"""
logger.info(f"Getting entities of type {entity_type}...")
# Use optimized label-based query from storage
raw_nodes = self.storage.get_nodes_by_label(graph_id, entity_type)
result = []
for node in raw_nodes:
result.append(NodeInfo(
uuid=node.get("uuid", ""),
name=node.get("name", ""),
labels=node.get("labels", []),
summary=node.get("summary", ""),
attributes=node.get("attributes", {})
))
logger.info(f"Found {len(result)} entities of type {entity_type}")
return result
def get_entity_summary(
self,
graph_id: str,
entity_name: str
) -> Dict[str, Any]:
"""Get relationship summary for a specific entity"""
logger.info(f"Getting relationship summary for entity {entity_name}...")
search_result = self.search_graph(
graph_id=graph_id,
query=entity_name,
limit=20
)
all_nodes = self.get_all_nodes(graph_id)
entity_node = None
for node in all_nodes:
if node.name.lower() == entity_name.lower():
entity_node = node
break
related_edges = []
if entity_node:
related_edges = self.get_node_edges(graph_id, entity_node.uuid)
return {
"entity_name": entity_name,
"entity_info": entity_node.to_dict() if entity_node else None,
"related_facts": search_result.facts,
"related_edges": [e.to_dict() for e in related_edges],
"total_relations": len(related_edges)
}
def get_graph_statistics(self, graph_id: str) -> Dict[str, Any]:
"""Get statistics for the graph"""
logger.info(f"Getting statistics for graph {graph_id}...")
nodes = self.get_all_nodes(graph_id)
edges = self.get_all_edges(graph_id)
entity_types = {}
for node in nodes:
for label in node.labels:
if label not in ["Entity", "Node"]:
entity_types[label] = entity_types.get(label, 0) + 1
relation_types = {}
for edge in edges:
relation_types[edge.name] = relation_types.get(edge.name, 0) + 1
return {
"graph_id": graph_id,
"total_nodes": len(nodes),
"total_edges": len(edges),
"entity_types": entity_types,
"relation_types": relation_types
}
def get_simulation_context(
self,
graph_id: str,
simulation_requirement: str,
limit: int = 30
) -> Dict[str, Any]:
"""Get simulation-related context information"""
logger.info(f"Getting simulation context: {simulation_requirement[:50]}...")
search_result = self.search_graph(
graph_id=graph_id,
query=simulation_requirement,
limit=limit
)
stats = self.get_graph_statistics(graph_id)
all_nodes = self.get_all_nodes(graph_id)
entities = []
for node in all_nodes:
custom_labels = [la for la in node.labels if la not in ["Entity", "Node"]]
if custom_labels:
entities.append({
"name": node.name,
"type": custom_labels[0],
"summary": node.summary
})
return {
"simulation_requirement": simulation_requirement,
"related_facts": search_result.facts,
"graph_statistics": stats,
"entities": entities[:limit],
"total_entities": len(entities)
}
# ========== Core Retrieval Tools (Optimized) ==========
def insight_forge(
self,
graph_id: str,
query: str,
simulation_requirement: str,
report_context: str = "",
max_sub_queries: int = 5
) -> InsightForgeResult:
"""
[InsightForge - Deep Insight Retrieval]
The most powerful hybrid retrieval function, automatically decomposes problems and performs multi-dimensional retrieval:
1. Use LLM to decompose the problem into multiple sub-questions
2. Perform semantic search on each sub-question
3. Extract related entities and get their detailed information
4. Trace relationship chains
5. Integrate all results and generate deep insights
"""
logger.info(f"InsightForge deep insight retrieval: {query[:50]}...")
result = InsightForgeResult(
query=query,
simulation_requirement=simulation_requirement,
sub_queries=[]
)
# Step 1: Use LLM to generate sub-questions
sub_queries = self._generate_sub_queries(
query=query,
simulation_requirement=simulation_requirement,
report_context=report_context,
max_queries=max_sub_queries
)
result.sub_queries = sub_queries
logger.info(f"Generated {len(sub_queries)} sub-questions")
# Step 2: Perform semantic search on each sub-question
all_facts = []
all_edges = []
seen_facts = set()
for sub_query in sub_queries:
search_result = self.search_graph(
graph_id=graph_id,
query=sub_query,
limit=15,
scope="edges"
)
for fact in search_result.facts:
if fact not in seen_facts:
all_facts.append(fact)
seen_facts.add(fact)
all_edges.extend(search_result.edges)
# Also search for the original question
main_search = self.search_graph(
graph_id=graph_id,
query=query,
limit=20,
scope="edges"
)
for fact in main_search.facts:
if fact not in seen_facts:
all_facts.append(fact)
seen_facts.add(fact)
result.semantic_facts = all_facts
result.total_facts = len(all_facts)
# Step 3: Extract related entity UUIDs from edges
entity_uuids = set()
for edge_data in all_edges:
if isinstance(edge_data, dict):
source_uuid = edge_data.get('source_node_uuid', '')
target_uuid = edge_data.get('target_node_uuid', '')
if source_uuid:
entity_uuids.add(source_uuid)
if target_uuid:
entity_uuids.add(target_uuid)
# Get related entity details
entity_insights = []
node_map = {}
for uuid in list(entity_uuids):
if not uuid:
continue
try:
node = self.get_node_detail(uuid)
if node:
node_map[uuid] = node
entity_type = next((la for la in node.labels if la not in ["Entity", "Node"]), "Entity")
related_facts = [
f for f in all_facts
if node.name.lower() in f.lower()
]
entity_insights.append({
"uuid": node.uuid,
"name": node.name,
"type": entity_type,
"summary": node.summary,
"related_facts": related_facts
})
except Exception as e:
logger.debug(f"Failed to get node {uuid}: {e}")
continue
result.entity_insights = entity_insights
result.total_entities = len(entity_insights)
# Step 4: Build relationship chains
relationship_chains = []
for edge_data in all_edges:
if isinstance(edge_data, dict):
source_uuid = edge_data.get('source_node_uuid', '')
target_uuid = edge_data.get('target_node_uuid', '')
relation_name = edge_data.get('name', '')
source_name = node_map.get(source_uuid, NodeInfo('', '', [], '', {})).name or source_uuid[:8]
target_name = node_map.get(target_uuid, NodeInfo('', '', [], '', {})).name or target_uuid[:8]
chain = f"{source_name} --[{relation_name}]--> {target_name}"
if chain not in relationship_chains:
relationship_chains.append(chain)
result.relationship_chains = relationship_chains
result.total_relationships = len(relationship_chains)
logger.info(f"InsightForge complete: {result.total_facts} facts, {result.total_entities} entities, {result.total_relationships} relationships")
return result
def _generate_sub_queries(
self,
query: str,
simulation_requirement: str,
report_context: str = "",
max_queries: int = 5
) -> List[str]:
"""Use LLM to generate sub-questions"""
system_prompt = """You are a professional question analysis expert. Your task is to decompose a complex question into multiple sub-questions that can be independently observed in a simulated world.
Requirements:
1. Each sub-question should be specific enough to find related Agent behavior or events in the simulated world
2. Sub-questions should cover different dimensions of the original question (e.g., who, what, why, how, when, where)
3. Sub-questions should be relevant to the simulation scenario
4. Return in JSON format: {"sub_queries": ["sub-question 1", "sub-question 2", ...]}"""
user_prompt = f"""Simulation requirement background:
{simulation_requirement}
{f"Report context: {report_context[:500]}" if report_context else ""}
Please decompose the following question into {max_queries} sub-questions:
{query}
Return the sub-questions as a JSON list."""
try:
response = self.llm.chat_json(
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
temperature=0.3
)
sub_queries = response.get("sub_queries", [])
return [str(sq) for sq in sub_queries[:max_queries]]
except Exception as e:
logger.warning(f"Failed to generate sub-questions: {str(e)}, using default sub-questions")
return [
query,
f"Main participants in {query}",
f"Causes and impacts of {query}",
f"Development process of {query}"
][:max_queries]
def panorama_search(
self,
graph_id: str,
query: str,
include_expired: bool = True,
limit: int = 50
) -> PanoramaResult:
"""
[PanoramaSearch - Breadth Search]
Get a comprehensive panoramic view, including all related content and historical/expired information.
"""
logger.info(f"PanoramaSearch breadth search: {query[:50]}...")
result = PanoramaResult(query=query)
# Get all nodes
all_nodes = self.get_all_nodes(graph_id)
node_map = {n.uuid: n for n in all_nodes}
result.all_nodes = all_nodes
result.total_nodes = len(all_nodes)
# Get all edges (including temporal information)
all_edges = self.get_all_edges(graph_id, include_temporal=True)
result.all_edges = all_edges
result.total_edges = len(all_edges)
# Categorize facts
active_facts = []
historical_facts = []
for edge in all_edges:
if not edge.fact:
continue
source_name = node_map.get(edge.source_node_uuid, NodeInfo('', '', [], '', {})).name or edge.source_node_uuid[:8]
target_name = node_map.get(edge.target_node_uuid, NodeInfo('', '', [], '', {})).name or edge.target_node_uuid[:8]
is_historical = edge.is_expired or edge.is_invalid
if is_historical:
valid_at = edge.valid_at or "Unknown"
invalid_at = edge.invalid_at or edge.expired_at or "Unknown"
fact_with_time = f"[{valid_at} - {invalid_at}] {edge.fact}"
historical_facts.append(fact_with_time)
else:
active_facts.append(edge.fact)
# Sort by relevance based on query
query_lower = query.lower()
keywords = [w.strip() for w in query_lower.replace(',', ' ').replace('，', ' ').split() if len(w.strip()) > 1]
def relevance_score(fact: str) -> int:
fact_lower = fact.lower()
score = 0
if query_lower in fact_lower:
score += 100
for kw in keywords:
if kw in fact_lower:
score += 10
return score
active_facts.sort(key=relevance_score, reverse=True)
historical_facts.sort(key=relevance_score, reverse=True)
result.active_facts = active_facts[:limit]
result.historical_facts = historical_facts[:limit] if include_expired else []
result.active_count = len(active_facts)
result.historical_count = len(historical_facts)
logger.info(f"PanoramaSearch complete: {result.active_count} valid, {result.historical_count} historical")
return result
def quick_search(
self,
graph_id: str,
query: str,
limit: int = 10
) -> SearchResult:
"""
[QuickSearch - Simple Search]
Fast and lightweight retrieval tool.
"""
logger.info(f"QuickSearch simple search: {query[:50]}...")
result = self.search_graph(
graph_id=graph_id,
query=query,
limit=limit,
scope="edges"
)
logger.info(f"QuickSearch complete: {result.total_count} results")
return result
def interview_agents(
self,
simulation_id: str,
interview_requirement: str,
simulation_requirement: str = "",
max_agents: int = 5,
custom_questions: List[str] = None
) -> InterviewResult:
"""
[InterviewAgents - Deep Interview]
Call the real OASIS interview API to interview Agents running in the simulation.
This method does NOT use GraphStorage — it calls SimulationRunner
and reads agent profiles from disk.
"""
from .simulation_runner import SimulationRunner
logger.info(f"InterviewAgents deep interview (real API): {interview_requirement[:50]}...")
result = InterviewResult(
interview_topic=interview_requirement,
interview_questions=custom_questions or []
)
# Step 1: Read agent profile files
profiles = self._load_agent_profiles(simulation_id)
if not profiles:
logger.warning(f"No profile files found for simulation {simulation_id}")
result.summary = "No Agent profile files found for interview"
return result
result.total_agents = len(profiles)
logger.info(f"Loaded {len(profiles)} Agent profiles")
# Step 2: Use LLM to select Agents for interview
selected_agents, selected_indices, selection_reasoning = self._select_agents_for_interview(
profiles=profiles,
interview_requirement=interview_requirement,
simulation_requirement=simulation_requirement,
max_agents=max_agents
)
result.selected_agents = selected_agents
result.selection_reasoning = selection_reasoning
logger.info(f"Selected {len(selected_agents)} Agents for interview: {selected_indices}")
# Step 3: Generate interview questions
if not result.interview_questions:
result.interview_questions = self._generate_interview_questions(
interview_requirement=interview_requirement,
simulation_requirement=simulation_requirement,
selected_agents=selected_agents
)
logger.info(f"Generated {len(result.interview_questions)} interview questions")
combined_prompt = "\n".join([f"{i+1}. {q}" for i, q in enumerate(result.interview_questions)])
INTERVIEW_PROMPT_PREFIX = (
"You are being interviewed. Please combine your character profile, all past memories and actions, "
"and directly answer the following questions in plain text.\n"
"Response requirements:\n"
"1. Answer directly in natural language, do not call any tools\n"
"2. Do not return JSON format or tool call format\n"
"3. Do not use Markdown headings (e.g., #, ##, ###)\n"
"4. Answer the questions in order, with each answer starting with 'Question X:' (X is the question number)\n"
"5. Separate each answer with a blank line\n"
"6. Provide substantive answers, at least 2-3 sentences per question\n\n"
)
optimized_prompt = f"{INTERVIEW_PROMPT_PREFIX}{combined_prompt}"
# Step 4: Call the real interview API
try:
interviews_request = []
for agent_idx in selected_indices:
interviews_request.append({
"agent_id": agent_idx,
"prompt": optimized_prompt
})
logger.info(f"Calling batch interview API (dual platform): {len(interviews_request)} Agents")
api_result = SimulationRunner.interview_agents_batch(
simulation_id=simulation_id,
interviews=interviews_request,
platform=None,
timeout=180.0
)
logger.info(f"Interview API returned: {api_result.get('interviews_count', 0)} results, success={api_result.get('success')}")
if not api_result.get("success", False):
error_msg = api_result.get("error", "Unknown error")
logger.warning(f"Interview API call failed: {error_msg}")
result.summary = f"Interview API call failed: {error_msg}. Please check the OASIS simulation environment status."
return result
# Step 5: Parse API response
api_data = api_result.get("result", {})
results_dict = api_data.get("results", {}) if isinstance(api_data, dict) else {}
for i, agent_idx in enumerate(selected_indices):
agent = selected_agents[i]
agent_name = agent.get("realname", agent.get("username", f"Agent_{agent_idx}"))
agent_role = agent.get("profession", "Unknown")
agent_bio = agent.get("bio", "")
twitter_result = results_dict.get(f"twitter_{agent_idx}", {})
reddit_result = results_dict.get(f"reddit_{agent_idx}", {})
twitter_response = twitter_result.get("response", "")
reddit_response = reddit_result.get("response", "")
twitter_response = self._clean_tool_call_response(twitter_response)
reddit_response = self._clean_tool_call_response(reddit_response)
twitter_text = twitter_response if twitter_response else "(No response from this platform)"
reddit_text = reddit_response if reddit_response else "(No response from this platform)"
response_text = f"[Twitter Platform Response]\n{twitter_text}\n\n[Reddit Platform Response]\n{reddit_text}"
import re
combined_responses = f"{twitter_response} {reddit_response}"
clean_text = re.sub(r'#{1,6}\s+', '', combined_responses)
clean_text = re.sub(r'\{[^}]*tool_name[^}]*\}', '', clean_text)
clean_text = re.sub(r'[*_`|>~\-]{2,}', '', clean_text)
clean_text = re.sub(r'Question\d+[：:]\s*', '', clean_text)
clean_text = re.sub(r'【[^】]+】', '', clean_text)
sentences = re.split(r'[。！？]', clean_text)
meaningful = [
s.strip() for s in sentences
if 20 <= len(s.strip()) <= 150
and not re.match(r'^[\s\W，,；;：:、]+', s.strip())
and not s.strip().startswith(('{', 'Question'))
]
meaningful.sort(key=len, reverse=True)
key_quotes = [s + "。" for s in meaningful[:3]]
if not key_quotes:
paired = re.findall(r'\u201c([^\u201c\u201d]{15,100})\u201d', clean_text)
paired += re.findall(r'\u300c([^\u300c\u300d]{15,100})\u300d', clean_text)
key_quotes = [q for q in paired if not re.match(r'^[，,；;：:、]', q)][:3]
interview = AgentInterview(
agent_name=agent_name,
agent_role=agent_role,
agent_bio=agent_bio[:1000],
question=combined_prompt,
response=response_text,
key_quotes=key_quotes[:5]
)
result.interviews.append(interview)
result.interviewed_count = len(result.interviews)
except ValueError as e:
logger.warning(f"Interview API call failed (environment not running?): {e}")
result.summary = f"Interview failed: {str(e)}. The simulation environment may be closed. Please ensure the OASIS environment is running."
return result
except Exception as e:
logger.error(f"Interview API call exception: {e}")
import traceback
logger.error(traceback.format_exc())
result.summary = f"An error occurred during the interview process: {str(e)}"
return result
# Step 6: Generate interview summary
if result.interviews:
result.summary = self._generate_interview_summary(
interviews=result.interviews,
interview_requirement=interview_requirement
)
logger.info(f"InterviewAgents complete: Interviewed {result.interviewed_count} Agents (dual platform)")
return result
@staticmethod
def _clean_tool_call_response(response: str) -> str:
"""Clean JSON tool call wrappers in Agent responses and extract actual content"""
if not response or not response.strip().startswith('{'):
return response
text = response.strip()
if 'tool_name' not in text[:80]:
return response
import re as _re
try:
data = json.loads(text)
if isinstance(data, dict) and 'arguments' in data:
for key in ('content', 'text', 'body', 'message', 'reply'):
if key in data['arguments']:
return str(data['arguments'][key])
except (json.JSONDecodeError, KeyError, TypeError):
match = _re.search(r'"content"\s*:\s*"((?:[^"\\]|\\.)*)"', text)
if match:
return match.group(1).replace('\\n', '\n').replace('\\"', '"')
return response
def _load_agent_profiles(self, simulation_id: str) -> List[Dict[str, Any]]:
"""Load Agent profile files for simulation"""
import os
import csv
sim_dir = os.path.join(
os.path.dirname(__file__),
f'../../uploads/simulations/{simulation_id}'
)
profiles = []
# Preferentially try to read Reddit JSON format
reddit_profile_path = os.path.join(sim_dir, "reddit_profiles.json")
if os.path.exists(reddit_profile_path):
try:
with open(reddit_profile_path, 'r', encoding='utf-8') as f:
profiles = json.load(f)
logger.info(f"Loaded {len(profiles)} profiles from reddit_profiles.json")
return profiles
except Exception as e:
logger.warning(f"Failed to read reddit_profiles.json: {e}")
# Try to read Twitter CSV format
twitter_profile_path = os.path.join(sim_dir, "twitter_profiles.csv")
if os.path.exists(twitter_profile_path):
try:
with open(twitter_profile_path, 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row in reader:
profiles.append({
"realname": row.get("name", ""),
"username": row.get("username", ""),
"bio": row.get("description", ""),
"persona": row.get("user_char", ""),
"profession": "Unknown"
})
logger.info(f"Loaded {len(profiles)} profiles from twitter_profiles.csv")
return profiles
except Exception as e:
logger.warning(f"Failed to read twitter_profiles.csv: {e}")
return profiles
def _select_agents_for_interview(
self,
profiles: List[Dict[str, Any]],
interview_requirement: str,
simulation_requirement: str,
max_agents: int
) -> tuple:
"""Use LLM to select Agents for interview"""
agent_summaries = []
for i, profile in enumerate(profiles):
summary = {
"index": i,
"name": profile.get("realname", profile.get("username", f"Agent_{i}")),
"profession": profile.get("profession", "Unknown"),
"bio": profile.get("bio", "")[:200],
"interested_topics": profile.get("interested_topics", [])
}
agent_summaries.append(summary)
system_prompt = """You are a professional interview planning expert. Your task is to select the most suitable Agents for interview from the simulated Agent list based on the interview requirements.
Selection Criteria:
1. Agent's identity/profession is relevant to the interview topic
2. Agent may hold unique or valuable perspectives
3. Select diverse perspectives (e.g., supporters, opposers, neutral, experts, etc.)
4. Prioritize roles directly related to the event
Return JSON format:
{
"selected_indices": [List of indices of selected Agents],
"reasoning": "Explanation of selection rationale"
}"""
user_prompt = f"""Interview Requirement:
{interview_requirement}
Simulation Background:
{simulation_requirement if simulation_requirement else "Not provided"}
Available Agent List ({len(agent_summaries)} total):
{json.dumps(agent_summaries, ensure_ascii=False, indent=2)}
Please select up to {max_agents} most suitable Agents for interview and explain your selection rationale."""
try:
response = self.llm.chat_json(
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
temperature=0.3
)
selected_indices = response.get("selected_indices", [])[:max_agents]
reasoning = response.get("reasoning", "Automatically selected based on relevance")
selected_agents = []
valid_indices = []
for idx in selected_indices:
if 0 <= idx < len(profiles):
selected_agents.append(profiles[idx])
valid_indices.append(idx)
return selected_agents, valid_indices, reasoning
except Exception as e:
logger.warning(f"LLM agent selection failed, using default selection: {e}")
selected = profiles[:max_agents]
indices = list(range(min(max_agents, len(profiles))))
return selected, indices, "Using default selection strategy"
def _generate_interview_questions(
self,
interview_requirement: str,
simulation_requirement: str,
selected_agents: List[Dict[str, Any]]
) -> List[str]:
"""Use LLM to generate interview questions"""
agent_roles = [a.get("profession", "Unknown") for a in selected_agents]
system_prompt = """You are a professional journalist/interviewer. Based on the interview requirements, generate 3-5 deep interview questions.
Question Requirements:
1. Open-ended questions that encourage detailed answers
2. Questions that may have different answers for different roles
3. Cover multiple dimensions: facts, viewpoints, feelings, etc.
4. Natural language, like real interviews
5. Keep each question under 50 characters, concise and clear
6. Ask directly, do not include background explanation or prefix
Return JSON format: {"questions": ["question1", "question2", ...]}"""
user_prompt = f"""Interview Requirement: {interview_requirement}
Simulation Background: {simulation_requirement if simulation_requirement else "Not provided"}
Interview Subject Roles: {', '.join(agent_roles)}
Please generate 3-5 interview questions."""
try:
response = self.llm.chat_json(
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
temperature=0.5
)
return response.get("questions", [f"What is your perspective on {interview_requirement}?"])
except Exception as e:
logger.warning(f"Failed to generate interview questions: {e}")
return [
f"What is your perspective on {interview_requirement}?",
"What impact does this have on you or the group you represent?",
"How do you think this issue should be solved or improved?"
]
def _generate_interview_summary(
self,
interviews: List[AgentInterview],
interview_requirement: str
) -> str:
"""Generate interview summary"""
if not interviews:
return "No interviews completed"
interview_texts = []
for interview in interviews:
interview_texts.append(f"[{interview.agent_name} ({interview.agent_role})]\n{interview.response[:500]}")
system_prompt = """You are a professional news editor. Please generate an interview summary based on the responses from multiple interviewees.
Summary Requirements:
1. Extract main viewpoints from all parties
2. Point out consensus and disagreement among viewpoints
3. Highlight valuable quotes
4. Remain objective and neutral, do not favor any side
5. Keep it under 1000 words
Format Constraints (Must Follow):
- Use plain text paragraphs, separated by blank lines
- Do not use Markdown headings (e.g., #, ##, ###)
- Do not use dividers (e.g., ---, ***)
- Use appropriate quotes when citing interviewees
- Can use **bold** to mark keywords, but do not use other Markdown syntax"""
user_prompt = f"""Interview Topic: {interview_requirement}
Interview Content:
{"".join(interview_texts)}
Please generate an interview summary."""
try:
summary = self.llm.chat(
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
temperature=0.3,
max_tokens=800
)
return summary
except Exception as e:
logger.warning(f"Failed to generate interview summary: {e}")
return f"Interviewed {len(interviews)} interviewees, including: " + ", ".join([i.agent_name for i in interviews])