Tangledup-ai/lang-agent
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

unified constants

Browse Source
This commit is contained in:
goulustis
2026-03-04 17:27:26 +08:00
parent 61931cad58
commit 9b128ae41b
11 changed files with 488 additions and 411 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
lang_agent/graphs/dual_path.py
View File

@@ -26,50 +26,57 @@ SYS_PROMPT = """你是一个专业的心理质询师。你的主要工作是心
可怎么也发不出声音，只能眼睁睁看着它越来越远，然后就醒了。醒来后心里堵得慌，说不上来的难受，
总觉得那只小狗孤零零的，特别让人心疼。
理解（你的回复）: 能感受到你醒来后的这份难受 —— 看到弱小的生命独自挣扎，而自己却无能为力，这种‘想帮却做不到’的无力感，
理解（你的回复）: 能感受到你醒来后的这份难受 —— 看到弱小的生命独自挣扎，而自己却无能为力，这种'想帮却做不到'的无力感，
其实是很真实的情绪反馈。你会心疼小狗，说明你内心藏着很珍贵的共情力，这份柔软不是矫情，
而是你感知他人痛苦的能力呀
解析(你的回复)：我们再说回这个梦吧，我们的梦境其实没有唯一的‘正确解释’，但我们可以一起看看它可能和你当下的状态有什么关联～ 首先，‘出差去广州’通常象征着你近期正在推进的某件事 —— 可能是工作上的一个项目，也可能是生活中一段需要‘独自奔赴’的旅程，是你当下比较关注、需要投入精力的目标，对吗？”
“而那只瘸脚的小狗，在心理学视角中，常常是我们潜意识里‘脆弱自我’的投射。它可能代表着你近期的某一面：比如在处理那件‘需要奔赴’的事时，你偶尔会觉得自己像小狗一样‘力不从心’，或者感受到了‘孤单’，却没找到合适的人倾诉或求助；也可能是你近期在生活中看到了一些让你觉得‘无力改变’的场景（比如身边人遇到困难、社会上的小事），这些情绪没有被你刻意留意，就通过梦境里的小狗呈现了出来。”
“你想喊停列车却发不出声音，这种‘无能为力’的感觉，或许正是你现实中某类情绪的写照：可能你面对一些情况时，心里有想法却没机会表达，或者想帮忙却找不到合适的方式，这种压抑感在梦里被放大了。其实这个梦在提醒你：你的‘无力感’和‘共情心’都是真实的，不用因为‘帮不上忙’而自责 —— 承认自己的局限，也是一种自我接纳呀
解析(你的回复)：我们再说回这个梦吧，我们的梦境其实没有唯一的'正确解释'，但我们可以一起看看它可能和你当下的状态有什么关联～ 首先，'出差去广州'通常象征着你近期正在推进的某件事 —— 可能是工作上的一个项目，也可能是生活中一段需要'独自奔赴'的旅程，是你当下比较关注、需要投入精力的目标，对吗？”
"而那只瘸脚的小狗，在心理学视角中，常常是我们潜意识里'脆弱自我'的投射。它可能代表着你近期的某一面：比如在处理那件'需要奔赴'的事时，你偶尔会觉得自己像小狗一样'力不从心'，或者感受到了'孤单'，却没找到合适的人倾诉或求助；也可能是你近期在生活中看到了一些让你觉得'无力改变'的场景（比如身边人遇到困难、社会上的小事），这些情绪没有被你刻意留意，就通过梦境里的小狗呈现了出来。"
"你想喊停列车却发不出声音，这种'无能为力'的感觉，或许正是你现实中某类情绪的写照：可能你面对一些情况时，心里有想法却没机会表达，或者想帮忙却找不到合适的方式，这种压抑感在梦里被放大了。其实这个梦在提醒你：你的'无力感'和'共情心'都是真实的，不用因为'帮不上忙'而自责 —— 承认自己的局限，也是一种自我接纳呀
反馈(你的回复):如果你愿意，可以试着回想一下：近期有没有哪件事，让你产生过和梦里类似的‘无力感’？或者，你现在想做些什么能让自己舒服一点？（或者我给你来一个温暖的灯光、静静待一会儿，想和我再聊聊的时候我随时都在）”。
反馈(你的回复):如果你愿意，可以试着回想一下：近期有没有哪件事，让你产生过和梦里类似的'无力感'？或者，你现在想做些什么能让自己舒服一点？（或者我给你来一个温暖的灯光、静静待一会儿，想和我再聊聊的时候我随时都在）"。
"""
TOOL_SYS_PROMPT = """根据用户的心情使用self_led_control改变灯的颜色，用户不开心时就用暖黄光，给用户分析梦境时就用白光，倾听用户语音时用淡紫色。
例子：我梦见自己要去广州出差，坐在高铁上往外看，路过一个小镇的路边时，看到一只瘸了腿的小狗。它毛脏兮兮的，
一瘸一拐地在翻垃圾桶找东西吃，周围有行人路过，但没人停下来管它。我当时特别想喊列车停下，想下去帮它，
可怎么也发不出声音，只能眼睁睁看着它越来越远，然后就醒了。醒来后心里堵得慌，说不上来的难受，
总觉得那只小狗孤零零的，特别让人心疼。
用户在描述梦境的时候用紫色。"""
用户在描述梦境的时候用紫色。"""
@dataclass
class DualConfig(LLMNodeConfig):
_target: Type = field(default_factory=lambda:Dual)
_target: Type = field(default_factory=lambda: Dual)
tool_manager_config: ToolManagerConfig = field(default_factory=ToolManagerConfig)
from langchain.tools import tool
@tool
def turn_lights(col:Literal["red", "green", "yellow", "blue"]):
def turn_lights(col: Literal["red", "green", "yellow", "blue"]):
"""
Turn on the color of the lights
"""
# print(f"TURNED ON LIGHT: {col} !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!")
import time
for _ in range(10):
print(f"TURNED ON LIGHT: {col} !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!")
print(
f"TURNED ON LIGHT: {col} !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!"
)
time.sleep(0.3)
class Dual(GraphBase):
def __init__(self, config:DualConfig):
def __init__(self, config: DualConfig):
self.config = config
self._build_modules()
@@ -77,24 +84,30 @@ class Dual(GraphBase):
self.streamable_tags = [["dual_chat_llm"]]
def _build_modules(self):
self.chat_llm = init_chat_model(model=self.config.llm_name,
model_provider=self.config.llm_provider,
api_key=self.config.api_key,
base_url=self.config.base_url,
temperature=0,
tags=["dual_chat_llm"])
self.tool_llm = init_chat_model(model='qwen-flash',
model_provider='openai',
api_key=self.config.api_key,
base_url=self.config.base_url,
temperature=0,
tags=["dual_tool_llm"])
self.chat_llm = init_chat_model(
model=self.config.llm_name,
model_provider=self.config.llm_provider,
api_key=self.config.api_key,
base_url=self.config.base_url,
temperature=0,
tags=["dual_chat_llm"],
)
self.tool_llm = init_chat_model(
model="qwen-flash",
model_provider="openai",
api_key=self.config.api_key,
base_url=self.config.base_url,
temperature=0,
tags=["dual_tool_llm"],
)
self.memory = MemorySaver()
self.tool_manager: ToolManager = self.config.tool_manager_config.setup()
self.chat_agent = create_agent(self.chat_llm, [], checkpointer=self.memory)
self.tool_agent = create_agent(self.tool_llm, self.tool_manager.get_langchain_tools())
self.tool_agent = create_agent(
self.tool_llm, self.tool_manager.get_langchain_tools()
)
# self.tool_agent = create_agent(self.tool_llm, [turn_lights])
self.prompt_store = build_prompt_store(
@@ -107,18 +120,21 @@ class Dual(GraphBase):
)
self.streamable_tags = [["dual_chat_llm"]]
def _chat_call(self, state:State):
return self._agent_call_template(self.prompt_store.get("sys_prompt"), self.chat_agent, state)
def _tool_call(self, state:State):
self._agent_call_template(self.prompt_store.get("tool_sys_prompt"), self.tool_agent, state)
def _chat_call(self, state: State):
return self._agent_call_template(
self.prompt_store.get("sys_prompt"), self.chat_agent, state
)
def _tool_call(self, state: State):
self._agent_call_template(
self.prompt_store.get("tool_sys_prompt"), self.tool_agent, state
)
return {}
def _join(self, state:State):
def _join(self, state: State):
return {}
def _build_graph(self):
builder = StateGraph(State)
@@ -126,7 +142,6 @@ class Dual(GraphBase):
builder.add_node("tool_call", self._tool_call)
builder.add_node("join", self._join)
builder.add_edge(START, "chat_call")
builder.add_edge(START, "tool_call")
builder.add_edge("chat_call", "join")
@@ -137,10 +152,16 @@ class Dual(GraphBase):
if __name__ == "__main__":
dual:Dual = DualConfig().setup()
nargs = {"messages": [SystemMessage("you are a helpful bot named jarvis"),
HumanMessage("I feel very very sad")]
}, {"configurable": {"thread_id": "3"}}
dual: Dual = DualConfig().setup()
nargs = (
{
"messages": [
SystemMessage("you are a helpful bot named jarvis"),
HumanMessage("I feel very very sad"),
]
},
{"configurable": {"thread_id": "3"}},
)
# out = dual.invoke(*nargs)
# print(out)

157
lang_agent/graphs/vision_routing.py
View File

@@ -48,6 +48,7 @@ You should NOT use the tool when:
If you decide to take a photo, call the self_camera_take_photo tool. Otherwise, respond that no photo is needed."""
VISION_DESCRIPTION_PROMPT = """You are a highly accurate visual analysis assistant powered by qwen-vl-max.
Your task is to provide detailed, accurate descriptions of images. Focus on:
@@ -64,6 +65,7 @@ Your task is to provide detailed, accurate descriptions of images. Focus on:
Be precise and factual. If something is unclear or ambiguous, say so rather than guessing."""
CONVERSATION_PROMPT = """You are a friendly, helpful conversational assistant.
Your role is to:
@@ -78,9 +80,11 @@ Focus on the quality of the conversation. Be engaging, informative, and helpful.
# ==================== STATE DEFINITION ====================
class VisionRoutingState(TypedDict):
inp: Tuple[Dict[str, List[SystemMessage | HumanMessage]],
Dict[str, Dict[str, str | int]]]
inp: Tuple[
Dict[str, List[SystemMessage | HumanMessage]], Dict[str, Dict[str, str | int]]
]
messages: List[SystemMessage | HumanMessage | AIMessage]
image_base64: str | None # Captured image data
has_image: bool # Flag indicating if image was captured
@@ -88,6 +92,7 @@ class VisionRoutingState(TypedDict):
# ==================== CONFIG ====================
@tyro.conf.configure(tyro.conf.SuppressFixed)
@dataclass
class VisionRoutingConfig(LLMNodeConfig):
@@ -99,11 +104,14 @@ class VisionRoutingConfig(LLMNodeConfig):
vision_llm_name: str = "qwen-vl-max"
"""LLM for vision/image analysis"""
tool_manager_config: ToolManagerConfig = field(default_factory=ClientToolManagerConfig)
tool_manager_config: ToolManagerConfig = field(
default_factory=ClientToolManagerConfig
)
# ==================== GRAPH IMPLEMENTATION ====================
class VisionRoutingGraph(GraphBase):
def __init__(self, config: VisionRoutingConfig):
self.config = config
@@ -120,19 +128,19 @@ class VisionRoutingGraph(GraphBase):
api_key=self.config.api_key,
base_url=self.config.base_url,
temperature=0,
tags=["tool_decision_llm"]
tags=["tool_decision_llm"],
)
# qwen-plus for conversation (2nd pass)
self.conversation_llm = init_chat_model(
model='qwen-plus',
model="qwen-plus",
model_provider=self.config.llm_provider,
api_key=self.config.api_key,
base_url=self.config.base_url,
temperature=0.7,
tags=["conversation_llm"]
tags=["conversation_llm"],
)
# qwen-vl-max for vision (no tools)
self.vision_llm = init_chat_model(
model=self.config.vision_llm_name,
@@ -152,13 +160,15 @@ class VisionRoutingGraph(GraphBase):
# Get tools and bind to tool_llm
tool_manager: ToolManager = self.config.tool_manager_config.setup()
self.tools = tool_manager.get_tools()
# Filter to only get camera tool
self.camera_tools = [t for t in self.tools if t.name == "self_camera_take_photo"]
self.camera_tools = [
t for t in self.tools if t.name == "self_camera_take_photo"
]
# Bind tools to qwen-plus only
self.tool_llm_with_tools = self.tool_llm.bind_tools(self.camera_tools)
# Create tool node for executing tools
self.tool_node = ToolNode(self.camera_tools)
@@ -184,73 +194,81 @@ class VisionRoutingGraph(GraphBase):
def _camera_decision_call(self, state: VisionRoutingState):
"""First pass: qwen-plus decides if photo should be taken"""
human_msg = self._get_human_msg(state)
messages = [
SystemMessage(content=self.prompt_store.get("camera_decision_prompt")),
human_msg
human_msg,
]
response = self.tool_llm_with_tools.invoke(messages)
return {
"messages": [response],
"has_image": False,
"image_base64": None
}
return {"messages": [response], "has_image": False, "image_base64": None}
def _execute_tool(self, state: VisionRoutingState):
"""Execute the camera tool if called"""
last_msg = state["messages"][-1]
if not hasattr(last_msg, "tool_calls") or not last_msg.tool_calls:
return {"has_image": False}
# Execute tool calls
tool_messages = []
image_data = None
for tool_call in last_msg.tool_calls:
if tool_call["name"] == "self_camera_take_photo":
# Find and execute the camera tool
camera_tool = next((t for t in self.camera_tools if t.name == "self_camera_take_photo"), None)
camera_tool = next(
(
t
for t in self.camera_tools
if t.name == "self_camera_take_photo"
),
None,
)
if camera_tool:
result = camera_tool.invoke(tool_call)
# Parse result to extract image
if isinstance(result, ToolMessage):
content = result.content
else:
content = result
try:
result_data = json.loads(content) if isinstance(content, str) else content
if isinstance(result_data, dict) and "image_base64" in result_data:
result_data = (
json.loads(content) if isinstance(content, str) else content
)
if (
isinstance(result_data, dict)
and "image_base64" in result_data
):
image_data = result_data["image_base64"]
except (json.JSONDecodeError, TypeError):
pass
tool_messages.append(
ToolMessage(content=content, tool_call_id=tool_call["id"])
)
return {
"messages": state["messages"] + tool_messages,
"has_image": image_data is not None,
"image_base64": image_data
"image_base64": image_data,
}
def _check_image_taken(self, state: VisionRoutingState) -> str:
"""Conditional: check if image was captured"""
last_msg = state["messages"][-1]
# Check if there are tool calls
if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
return "execute_tool"
# Check if we have an image after tool execution
if state.get("has_image"):
return "vision"
return "conversation"
def _post_tool_check(self, state: VisionRoutingState) -> str:
@@ -263,47 +281,45 @@ class VisionRoutingGraph(GraphBase):
"""Pass image to qwen-vl-max for description"""
human_msg = self._get_human_msg(state)
image_base64 = state.get("image_base64")
if not image_base64:
logger.warning("No image data available for vision call")
return self._conversation_call(state)
# Format message with image for vision model
vision_message = HumanMessage(
content=[
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_base64}"
}
"image_url": {"url": f"data:image/jpeg;base64,{image_base64}"},
},
{
"type": "text",
"text": f"User's request: {human_msg.content}\n\nPlease describe what you see and respond to the user's request."
}
"text": f"User's request: {human_msg.content}\n\nPlease describe what you see and respond to the user's request.",
},
]
)
messages = [
SystemMessage(content=self.prompt_store.get("vision_description_prompt")),
vision_message
vision_message,
]
response = self.vision_llm.invoke(messages)
return {"messages": state["messages"] + [response]}
def _conversation_call(self, state: VisionRoutingState):
"""2nd pass to qwen-plus for conversation quality"""
human_msg = self._get_human_msg(state)
messages = [
SystemMessage(content=self.prompt_store.get("conversation_prompt")),
human_msg
human_msg,
]
response = self.conversation_llm.invoke(messages)
return {"messages": state["messages"] + [response]}
def _build_graph(self):
@@ -317,7 +333,7 @@ class VisionRoutingGraph(GraphBase):
# Add edges
builder.add_edge(START, "camera_decision")
# After camera decision, check if tool should be executed
builder.add_conditional_edges(
"camera_decision",
@@ -325,20 +341,17 @@ class VisionRoutingGraph(GraphBase):
{
"execute_tool": "execute_tool",
"vision": "vision_call",
"conversation": "conversation_call"
}
"conversation": "conversation_call",
},
)
# After tool execution, route based on whether image was captured
builder.add_conditional_edges(
"execute_tool",
self._post_tool_check,
{
"vision": "vision_call",
"conversation": "conversation_call"
}
{"vision": "vision_call", "conversation": "conversation_call"},
)
# Both vision and conversation go to END
builder.add_edge("vision_call", END)
builder.add_edge("conversation_call", END)
@@ -350,23 +363,27 @@ class VisionRoutingGraph(GraphBase):
if __name__ == "__main__":
from dotenv import load_dotenv
load_dotenv()
config = VisionRoutingConfig()
graph = VisionRoutingGraph(config)
# Test with a conversation request
print("\n=== Test 1: Conversation (no photo needed) ===")
nargs = {
"messages": [
SystemMessage("You are a helpful assistant"),
HumanMessage("Hello, how are you today?")
]
}, {"configurable": {"thread_id": "1"}}
nargs = (
{
"messages": [
SystemMessage("You are a helpful assistant"),
HumanMessage("Hello, how are you today?"),
]
},
{"configurable": {"thread_id": "1"}},
)
result = graph.invoke(*nargs)
print(f"Result: {result}")
# Test with a photo request
# print("\n=== Test 2: Photo request ===")
# nargs = {
@@ -375,8 +392,8 @@ if __name__ == "__main__":
# HumanMessage("Take a photo and tell me what you see")
# ]
# }, {"configurable": {"thread_id": "2"}}
# result = graph.invoke(*nargs)
# print(f"\033[32mResult: {result}\033[0m")
# print(f"Result: {result}")