)
20251212_搭建本地支持联网推理助手
一、问题
搭建本地助手，提升工作效率。
二、代码
importjsonimportosimportwarningsimportgradioasgrimportrequests AUTHORIZATION="Bearer XXXXXXXXXXX"PROXIES={'https':'http://XXXXXXXXXXX8080',# 替换为自己的域账号和密码'http':'http://XXXXXXXom:8080',}os.environ["no_proxy"]="localhost,127.0.0.1,::1"os.environ['GRADIO_ANALYTICS_ENABLED']='False'warnings.filterwarnings("ignore",category=requests.packages.urllib3.exceptions.InsecureRequestWarning)defstream_response_v3(question,history):messages=[]fornum,hisinenumerate(history):ifnum>=3:breakmessages.append({"role":"user","content":history[0]['content']})messages.append({"role":"assistant","content":history[1]['content']})passmessages.append({"role":"user","content":question})data={"model":"ep-20250718205753-s4hns","messages":messages,"stream":True}try:response=requests.post("https://ark.cn-beijing.volces.com/api/v3/chat/completions",headers={"Authorization":AUTHORIZATION,"Content-Type":"application/json"},proxies=PROXIES,verify=False,json=data,stream=True,timeout=30)response.raise_for_status()full_response=""forlineinresponse.iter_lines():ifline:decoded_line=line.decode('utf-8').lstrip('data: ').strip()ifdecoded_line=="[DONE]":breaktry:chunk=json.loads(decoded_line)ifchunk.get('choices'):delta=chunk['choices'][0].get('delta',{})ifdelta.get('content'):content=delta['content']full_response+=contentyieldfull_responseexceptjson.JSONDecodeError:continueexceptrequests.exceptions.RequestExceptionase:yieldf"请求发生错误：{str(e)}"exceptExceptionase:yieldf"处理响应时发生错误：{str(e)}"defchat_with_ark(message,history):""" 支持多轮对话的ARK聊天函数 Args: message: 用户当前输入的消息 history: 对话历史记录，格式为: [{'role': 'user/assistant', 'content': '消息内容'}, ...] Yields: 逐步生成的回复内容 """messages=[]forturninhistory:messages.append({"role":turn["role"],"content":[{"type":"input_text","text":turn["content"]}]})messages.append({"role":"user","content":[{"type":"input_text","text":message}]})payload={"model":"ep-20251210140512-99ld2","stream":True,"tools":[{"type":"web_search","max_keyword":3}],"input":messages# 使用包含历史的消息列表}response=requests.post("https://ark.cn-beijing.volces.com/api/v3/responses",headers={"Authorization":AUTHORIZATION,"Content-Type":"application/json"},json=payload,stream=True,verify=False,proxies=PROXIES)partial_response=""forlineinresponse.iter_lines():ifline:decoded_line=line.decode('utf-8')ifdecoded_line.startswith('data:'):try:data=json.loads(decoded_line[5:])ifdata.get("type")=="response.output_text.delta":delta=data.get("delta","")partial_response+=deltayieldpartial_responseexceptjson.JSONDecodeError:continuewithgr.Blocks()asdemo:withgr.Tabs():withgr.Tab("1. deepseek-v3-671b"):chatbot1=gr.Chatbot(height=580,type="messages")chat_interface1=gr.ChatInterface(fn=stream_response_v3,cache_examples=False,chatbot=chatbot1,fill_height=True)chat_interface1.scale=2examples1=gr.Examples(examples=["请帮我写python代码，功能要求如下：","请帮我修改python代码，功能要求如下：","下述错误是什么原因？怎么解决？","我的代码报如上，我的代码报错如下。错误原因是什么？怎么解决？",],inputs=chat_interface1.textbox,run_on_click=False,cache_examples=False,)withgr.Tab("2. deepseek-v3-671b(web search)"):chatbot2=gr.Chatbot(height=580,type="messages")chat_interface2=gr.ChatInterface(fn=chat_with_ark,cache_examples=False,chatbot=chatbot2,fill_height=True)chat_interface2.scale=2examples2=gr.Examples(examples=["请帮我写python代码，功能要求如下：","请帮我修改python代码，功能要求如下：","下述错误是什么原因？怎么解决？","我的代码报如上，我的代码报错如下。错误原因是什么？怎么解决？",],inputs=chat_interface2.textbox,run_on_click=False,cache_examples=False,)port=30005importpsutilforprocinpsutil.process_iter(['pid','name']):try:forconninproc.connections():ifconn.laddr.port==port:print(f"Killing process{proc.pid}({proc.name()})")proc.kill()breakexceptExceptionase:print(e)passdemo.queue().launch(server_name="127.0.0.1",server_port=port,share=False,show_error=True)
)